use http2 client for JPS cache

GitOrigin-RevId: b18913557c6ab31f1a52f77bdf09f79277374b74
2026-09-27 10:03:11 +07:00 · 2024-09-06 19:25:22 +00:00
parent c6cfc4db90
commit bfe323ec57
27 changed files with 1024 additions and 736 deletions
@@ -14,15 +14,19 @@ enum class AddDirEntriesMode {
  ALL
 }

+// `createFileParentDirs = false` can be used for performance reasons if you do create a lot of zip files
 fun zipWithCompression(
  targetFile: Path,
  dirs: Map<Path, String>,
  compressionLevel: Int = Deflater.DEFAULT_COMPRESSION,
  addDirEntriesMode: AddDirEntriesMode = AddDirEntriesMode.NONE,
  overwrite: Boolean = false,
+  createFileParentDirs: Boolean = true,
  fileFilter: ((name: String) -> Boolean)? = null,
 ) {
-  Files.createDirectories(targetFile.parent)
+  if (createFileParentDirs) {
+    Files.createDirectories(targetFile.parent)
+  }
  ZipFileWriter(
    channel = FileChannel.open(targetFile, if (overwrite) W_OVERWRITE else W_CREATE_NEW),
    deflater = if (compressionLevel == Deflater.NO_COMPRESSION) null else Deflater(compressionLevel, true),
@@ -22,8 +22,7 @@ abstract class BuildMessageLoggerBase : BuildMessageLogger() {
        Span.current().addEvent("artifact built: ${message.text}")
      }
      LogMessage.Kind.COMPILATION_ERRORS -> {
-        val errorsString = (message as CompilationErrorsLogMessage).errorMessages.joinToString(separator = "\n")
-        Span.current().addEvent("compilation errors (${message.compilerName}):\n$errorsString")
+        // reported as span event
      }
      LogMessage.Kind.BUILD_CANCEL, LogMessage.Kind.BUILD_PROBLEM -> {
        throw BuildScriptsLoggedError(message.text)
@@ -7,6 +7,7 @@ import kotlinx.collections.immutable.persistentListOf
 import kotlinx.collections.immutable.toPersistentMap
 import org.jetbrains.annotations.ApiStatus
 import org.jetbrains.annotations.TestOnly
+import org.jetbrains.intellij.build.BuildOptions.Companion.BUILD_STEPS_TO_SKIP_PROPERTY
 import org.jetbrains.intellij.build.BuildPaths.Companion.COMMUNITY_ROOT
 import org.jetbrains.intellij.build.dependencies.DependenciesProperties
 import org.jetbrains.intellij.build.dependencies.TeamCityHelper
@@ -15,6 +16,8 @@ import java.nio.file.Path
 import java.util.*
 import java.util.concurrent.TimeUnit
 import kotlin.random.Random
+import kotlin.time.Duration
+import kotlin.time.Duration.Companion.minutes

 data class BuildOptions(
  @ApiStatus.Internal @JvmField val jarCacheDir: Path? = null,
@@ -99,6 +102,11 @@ data class BuildOptions(
   * If `true`, the project modules will be compiled incrementally.
   */
  var incrementalCompilation: Boolean = getBooleanProperty(INTELLIJ_BUILD_INCREMENTAL_COMPILATION),
+
+  /**
+   * Full rebuild will be triggered if this timeout is exceeded for incremental compilation.
+   */
+  val incrementalCompilationTimeout: Duration? = System.getProperty("intellij.build.incremental.compilation.timeoutMin")?.toLong()?.minutes,
 ) {
  companion object {
    /**
@@ -385,11 +393,6 @@ data class BuildOptions(
   */
  var incrementalCompilationFallbackRebuild: Boolean = getBooleanProperty(INCREMENTAL_COMPILATION_FALLBACK_REBUILD_PROPERTY, true)

-  /**
-   * Full rebuild will be triggered if this timeout is exceeded for incremental compilation.
-   */
-  val incrementalCompilationTimeout: Long = System.getProperty("intellij.build.incremental.compilation.timeoutMin")?.toLong() ?: Long.MAX_VALUE
-
  /**
   * Use [BuildContext.buildNumber] to get the actual build number in build scripts.
   * @see BuildContext.reportDistributionBuildNumber
@@ -150,7 +150,7 @@ fun reorderJar(jarFile: Path, orderedNames: List<String>) {
        packageIndexBuilder.addFile(entry.name)
        zipCreator.uncompressedData(
          nameString = entry.name,
-          data = entry.getByteBuffer(sourceZip),
+          data = entry.getByteBuffer(sourceZip, null),
          indexWriter = packageIndexBuilder.indexWriter,
        )
      }
@@ -3,20 +3,16 @@

 package org.jetbrains.intellij.build.http2Client

-import io.netty.buffer.ByteBufUtil
 import io.netty.channel.ChannelHandlerContext
 import io.netty.handler.codec.http.HttpHeaderNames
-import io.netty.handler.codec.http.HttpHeaderValues
 import io.netty.handler.codec.http.HttpMethod
 import io.netty.handler.codec.http.HttpResponseStatus
 import io.netty.handler.codec.http2.Http2Headers
 import io.netty.handler.codec.http2.Http2HeadersFrame
 import io.netty.handler.codec.http2.ReadOnlyHttp2Headers
 import io.netty.util.AsciiString
-import kotlinx.serialization.DeserializationStrategy
-import kotlinx.serialization.serializer

-// https://cabulous.medium.com/http-2-and-how-it-works-9f645458e4b2
+// user-facing API on top of implementation API Http2ConnectionProvider
 internal class Http2ClientConnection internal constructor(
  private val scheme: AsciiString,
  private val authority: AsciiString,
@@ -33,30 +29,13 @@ internal class Http2ClientConnection internal constructor(
    connection.close()
  }

-  internal suspend inline fun <reified T : Any> post(path: String, data: String, contentType: AsciiString): T {
-    return post(path = AsciiString.of(path), data = data, contentType = contentType, deserializer = serializer<T>())
-  }
-
-  private suspend fun <T : Any> post(path: AsciiString, data: String, contentType: AsciiString, deserializer: DeserializationStrategy<T>): T {
-    return connection.stream { stream, result ->
-      val bufferAllocator = stream.alloc()
-      stream.pipeline().addLast(Http2StreamJsonInboundHandler(result = result, bufferAllocator = bufferAllocator, deserializer = deserializer))
-
-      stream.writeHeaders(
-        headers = ReadOnlyHttp2Headers.clientHeaders(
-          true,
-          HttpMethod.POST.asciiName(),
-          path,
-          scheme,
-          authority,
-          HttpHeaderNames.CONTENT_TYPE, contentType,
-          HttpHeaderNames.ACCEPT_ENCODING, HttpHeaderValues.GZIP,
-          *commonHeaders
-        ),
-        endStream = false,
-      )
-      stream.writeData(ByteBufUtil.writeUtf8(bufferAllocator, data), endStream = true)
-    }
+  fun withAuth(authHeader: CharSequence): Http2ClientConnection {
+    return Http2ClientConnection(
+      scheme = scheme,
+      authority = scheme,
+      commonHeaders = commonHeaders + arrayOf(HttpHeaderNames.AUTHORIZATION, AsciiString.of(authHeader)),
+      connection = connection,
+    )
  }

  suspend fun head(path: CharSequence): HttpResponseStatus {
@@ -92,8 +71,11 @@ internal class Http2ClientConnection internal constructor(
    }
  }

-
-  internal fun createHeaders(method: HttpMethod, path: AsciiString): Http2Headers {
-    return ReadOnlyHttp2Headers.clientHeaders(true, method.asciiName(), path, scheme, authority, *commonHeaders)
+  internal fun createHeaders(method: HttpMethod, path: AsciiString, vararg extraHeaders: AsciiString): Http2Headers {
+    var otherHeaders = commonHeaders
+    if (extraHeaders.isNotEmpty()) {
+      otherHeaders += extraHeaders
+    }
+    return ReadOnlyHttp2Headers.clientHeaders(true, method.asciiName(), path, scheme, authority, *otherHeaders)
  }
 }
@@ -62,7 +62,7 @@ internal suspend fun <T> withHttp2ClientConnectionFactory(
  useSsl: Boolean = true,
  trustAll: Boolean = false,
  block: suspend (Http2ClientConnectionFactory) -> T,
-) {
+): T {
  var connection: Http2ClientConnectionFactory? = null
  try {
    return supervisorScope {
@@ -10,6 +10,8 @@ import io.netty.channel.Channel
 import io.netty.channel.ChannelHandlerContext
 import io.netty.channel.ChannelInitializer
 import io.netty.channel.SimpleChannelInboundHandler
+import io.netty.handler.codec.http.HttpResponseStatus
+import io.netty.handler.codec.http.HttpStatusClass
 import io.netty.handler.codec.http2.*
 import io.netty.handler.ssl.SslContext
 import io.opentelemetry.api.common.AttributeKey
@@ -29,6 +31,8 @@ import kotlin.random.asJavaRandom
 import kotlin.time.Duration.Companion.milliseconds
 import kotlin.time.Duration.Companion.seconds

+internal class UnexpectedHttpStatus(@JvmField val status: HttpResponseStatus) : RuntimeException("Unexpected HTTP response status: $status")
+
 private class ConnectionState(
  @JvmField val bootstrap: Http2StreamChannelBootstrap,
  @JvmField val channel: Channel,
@@ -37,6 +41,7 @@ private class ConnectionState(

 private const val MAX_ATTEMPTS = 2

+// https://cabulous.medium.com/http-2-and-how-it-works-9f645458e4b2
 // https://stackoverflow.com/questions/55087292/how-to-handle-http-2-goaway-with-java-net-httpclient
 // Server can send GOAWAY frame after X streams, that's why we need manager for channel - open a new one in case of such error
 internal class Http2ConnectionProvider(
@@ -152,6 +157,12 @@ internal class Http2ConnectionProvider(
          continue
        }
      }
+      catch (e: UnexpectedHttpStatus) {
+        // retry only for sever errors
+        if (e.status.codeClass() != HttpStatusClass.SERVER_ERROR) {
+          throw e
+        }
+      }
      catch (e: Throwable) {
        if (attemptIndex >= MAX_ATTEMPTS) {
          if (suppressedExceptions != null) {
@@ -3,27 +3,101 @@ package org.jetbrains.intellij.build.http2Client

 import io.netty.buffer.ByteBufAllocator
 import io.netty.buffer.ByteBufInputStream
+import io.netty.buffer.ByteBufUtil
 import io.netty.channel.ChannelHandlerContext
 import io.netty.handler.codec.http.HttpHeaderNames
 import io.netty.handler.codec.http.HttpHeaderValues
+import io.netty.handler.codec.http.HttpMethod
 import io.netty.handler.codec.http.HttpResponseStatus
 import io.netty.handler.codec.http2.Http2DataFrame
 import io.netty.handler.codec.http2.Http2HeadersFrame
 import io.netty.handler.codec.http2.Http2StreamFrame
+import io.netty.util.AsciiString
 import kotlinx.coroutines.CompletableDeferred
 import kotlinx.serialization.DeserializationStrategy
+import kotlinx.serialization.KSerializer
 import kotlinx.serialization.json.Json
 import kotlinx.serialization.json.decodeFromStream
+import kotlinx.serialization.serializer
 import java.util.zip.GZIPInputStream

 private val json = Json {
  ignoreUnknownKeys = true
 }

-internal class Http2StreamJsonInboundHandler<T : Any>(
+internal suspend fun Http2ClientConnection.getString(path: CharSequence): String {
+  return connection.stream { stream, result ->
+    stream.pipeline().addLast(Http2StreamJsonInboundHandler(result = result, bufferAllocator = stream.alloc(), deserializer = null))
+
+    stream.writeHeaders(
+      headers = createHeaders(
+        HttpMethod.GET,
+        AsciiString.of(path),
+        HttpHeaderNames.ACCEPT_ENCODING, HttpHeaderValues.GZIP,
+      ),
+      endStream = true,
+    )
+  }
+}
+
+internal suspend inline fun <reified T : Any> Http2ClientConnection.getJsonOrDefaultIfNotFound(
+  path: CharSequence,
+  defaultIfNotFound: T,
+): T {
+  return doGetJsonOrDefaultIfNotFound(path = path, deserializer = serializer<T>(), defaultIfNotFound = defaultIfNotFound)
+}
+
+internal suspend inline fun <reified T : Any> Http2ClientConnection.post(path: CharSequence, data: CharSequence, contentType: AsciiString): T {
+  return post(path = AsciiString.of(path), data = data, contentType = contentType, deserializer = serializer<T>())
+}
+
+private suspend fun <T : Any> Http2ClientConnection.doGetJsonOrDefaultIfNotFound(
+  path: CharSequence,
+  deserializer: KSerializer<T>,
+  defaultIfNotFound: T,
+): T {
+  return connection.stream { stream, result ->
+    stream.pipeline().addLast(Http2StreamJsonInboundHandler(
+      result = result,
+      bufferAllocator = stream.alloc(),
+      deserializer = deserializer,
+      defaultIfNotFound = defaultIfNotFound,
+    ))
+
+    stream.writeHeaders(
+      headers = createHeaders(
+        HttpMethod.GET,
+        AsciiString.of(path),
+        HttpHeaderNames.ACCEPT_ENCODING, HttpHeaderValues.GZIP,
+      ),
+      endStream = true,
+    )
+  }
+}
+
+private suspend fun <T : Any> Http2ClientConnection.post(path: AsciiString, data: CharSequence, contentType: AsciiString, deserializer: DeserializationStrategy<T>): T {
+  return connection.stream { stream, result ->
+    val bufferAllocator = stream.alloc()
+    stream.pipeline().addLast(Http2StreamJsonInboundHandler(result = result, bufferAllocator = bufferAllocator, deserializer = deserializer))
+
+    stream.writeHeaders(
+      headers = createHeaders(
+        HttpMethod.POST,
+        path,
+        HttpHeaderNames.CONTENT_TYPE, contentType,
+        HttpHeaderNames.ACCEPT_ENCODING, HttpHeaderValues.GZIP,
+      ),
+      endStream = false,
+    )
+    stream.writeData(ByteBufUtil.writeUtf8(bufferAllocator, data), endStream = true)
+  }
+}
+
+private class Http2StreamJsonInboundHandler<T : Any>(
  private val bufferAllocator: ByteBufAllocator,
  private val result: CompletableDeferred<T>,
-  private val deserializer: DeserializationStrategy<T>,
+  private val deserializer: DeserializationStrategy<T>?,
+  private val defaultIfNotFound: T? = null,
 ) : InboundHandlerResultTracker<Http2StreamFrame>(result) {
  private var isGzip = false

@@ -52,18 +126,36 @@ internal class Http2StreamJsonInboundHandler<T : Any>(

      val response = if (isGzip) {
        GZIPInputStream(ByteBufInputStream(data)).use {
-          json.decodeFromStream(deserializer, it)
+          if (deserializer == null) {
+            @Suppress("UNCHECKED_CAST")
+            it.reader().readText() as T
+          }
+          else {
+            json.decodeFromStream(deserializer, it)
+          }
        }
      }
      else {
-        json.decodeFromString(deserializer, data.toString(Charsets.UTF_8))
+        val string = data.toString(Charsets.UTF_8)
+        if (deserializer == null) {
+          @Suppress("UNCHECKED_CAST")
+          string as T
+        }
+        else {
+          json.decodeFromString(deserializer, string)
+        }
      }
      result.complete(response)
    }
    else if (frame is Http2HeadersFrame) {
      val status = HttpResponseStatus.parseLine(frame.headers().status())
      if (status != HttpResponseStatus.OK) {
-        result.completeExceptionally(IllegalStateException("Failed with $status"))
+        if (status == HttpResponseStatus.NOT_FOUND && defaultIfNotFound != null) {
+          result.complete(defaultIfNotFound)
+        }
+        else {
+          result.completeExceptionally(UnexpectedHttpStatus(status))
+        }
        return
      }

@@ -24,6 +24,16 @@ private val OVERWRITE_OPERATION = EnumSet.of(StandardOpenOption.WRITE, StandardO

 internal data class DownloadResult(@JvmField var size: Long, @JvmField val digest: MessageDigest)

+internal suspend fun Http2ClientConnection.download(path: String, file: Path): Long {
+  Files.createDirectories(file.parent)
+
+  return connection.stream { stream, result ->
+    stream.pipeline().addLast(FileDownloadHandler(result = result, file = file))
+
+    stream.writeHeaders(createHeaders(HttpMethod.GET, AsciiString.of(path)), endStream = true)
+  }
+}
+
 internal suspend fun Http2ClientConnection.download(
  path: String,
  file: Path,
@@ -34,7 +44,7 @@ internal suspend fun Http2ClientConnection.download(

  return connection.stream { stream, result ->
    stream.pipeline().addLast(
-      DownloadHandler(
+      ZstdDecompressingFileDownloadHandler(
        result = result,
        downloadResult = DownloadResult(size = 0, digest = digestFactory()),
        file = file,
@@ -43,12 +53,10 @@ internal suspend fun Http2ClientConnection.download(
    )

    stream.writeHeaders(createHeaders(HttpMethod.GET, AsciiString.of(path)), endStream = true)
-
-    result.await()
  }
 }

-private class DownloadHandler(
+private class ZstdDecompressingFileDownloadHandler(
  private val result: CompletableDeferred<DownloadResult>,
  private val downloadResult: DownloadResult,
  private val file: Path,
@@ -82,7 +90,7 @@ private class DownloadHandler(
    if (frame is Http2HeadersFrame) {
      val status = HttpResponseStatus.parseLine(frame.headers().status())
      if (status != HttpResponseStatus.OK) {
-        result.completeExceptionally(IllegalStateException("Unexpected response status: $status"))
+        result.completeExceptionally(UnexpectedHttpStatus(status))
      }
    }
    else if (frame is Http2DataFrame) {
@@ -125,3 +133,50 @@ private class DownloadHandler(
    while (sourceBuffer.hasRemaining())
  }
 }
+
+private class FileDownloadHandler(
+  private val result: CompletableDeferred<Long>,
+  private val file: Path,
+) : InboundHandlerResultTracker<Http2StreamFrame>(result) {
+  private var offset = 0L
+  private var fileChannel: FileChannel? = null
+
+  override fun acceptInboundMessage(message: Any): Boolean = message is Http2DataFrame || message is Http2HeadersFrame
+
+  override fun handlerAdded(ctx: ChannelHandlerContext?) {
+    fileChannel = FileChannel.open(file, OVERWRITE_OPERATION)
+  }
+
+  override fun handlerRemoved(context: ChannelHandlerContext) {
+    fileChannel?.close()
+    fileChannel = null
+  }
+
+  override fun channelRead0(context: ChannelHandlerContext, frame: Http2StreamFrame) {
+    if (frame is Http2HeadersFrame) {
+      val status = HttpResponseStatus.parseLine(frame.headers().status())
+      if (status != HttpResponseStatus.OK) {
+        if (status == HttpResponseStatus.NOT_FOUND) {
+          result.complete(-1)
+        }
+        else {
+          result.completeExceptionally(UnexpectedHttpStatus(status))
+        }
+      }
+    }
+    else if (frame is Http2DataFrame) {
+      val content = frame.content()
+      val fileChannel = fileChannel!!
+      while (true) {
+        val readableBytes = content.readableBytes()
+        if (readableBytes == 0) {
+          break
+        }
+        offset += content.readBytes(fileChannel, offset, readableBytes)
+      }
+      if (frame.isEndStream) {
+        result.complete(offset)
+      }
+    }
+  }
+}
@@ -1,9 +1,13 @@
 // Copyright 2000-2024 JetBrains s.r.o. and contributors. Use of this source code is governed by the Apache 2.0 license.
+@file:Suppress("DuplicatedCode")
+
 package org.jetbrains.intellij.build.http2Client

 import com.github.luben.zstd.Zstd
 import com.github.luben.zstd.ZstdCompressCtx
 import io.netty.channel.ChannelHandlerContext
+import io.netty.handler.codec.http.HttpHeaderNames
+import io.netty.handler.codec.http.HttpHeaderValues
 import io.netty.handler.codec.http.HttpMethod
 import io.netty.handler.codec.http.HttpResponseStatus
 import io.netty.handler.codec.http2.DefaultHttp2DataFrame
@@ -20,39 +24,54 @@ import java.util.*
 import kotlin.math.min

 internal val READ_OPERATION = EnumSet.of(StandardOpenOption.READ)
+
 internal data class UploadResult(@JvmField var uploadedSize: Long, @JvmField var fileSize: Long)

+internal suspend fun Http2ClientConnection.upload(path: CharSequence, file: Path): UploadResult {
+  return upload(path = path, file = file, sourceBlockSize = 1024 * 1024, zstdCompressContextPool = null)
+}
+
 internal suspend fun Http2ClientConnection.upload(
-  path: AsciiString,
+  path: CharSequence,
  file: Path,
  sourceBlockSize: Int,
-  zstdCompressContextPool: ZstdCompressContextPool,
+  zstdCompressContextPool: ZstdCompressContextPool?,
 ): UploadResult {
  return connection.stream { stream, result ->
    val handler = WebDavPutStatusChecker(result)
    stream.pipeline().addLast(handler)

-    stream.writeHeaders(createHeaders(HttpMethod.PUT, path), endStream = false)
+    stream.writeHeaders(createHeaders(HttpMethod.PUT, AsciiString.of(path), HttpHeaderNames.CONTENT_TYPE, HttpHeaderValues.APPLICATION_OCTET_STREAM), endStream = false)

-    val fileBuffer = FileChannel.open(file, READ_OPERATION).use { channel ->
-      channel.map(FileChannel.MapMode.READ_ONLY, 0, channel.size())
-    }
-    try {
-      val fileSize = fileBuffer.remaining().toLong()
-      zstdCompressContextPool.withZstd(fileSize) { zstd ->
-        compressAndUpload(
-          fileBuffer = fileBuffer,
-          sourceBlockSize = sourceBlockSize,
-          zstd = zstd,
-          stream = stream,
-          fileSize = fileSize,
-        ) {
+    if (zstdCompressContextPool == null) {
+      FileChannel.open(file, READ_OPERATION).use { channel ->
+        uploadUncompressed(fileChannel = channel, sourceBlockSize = sourceBlockSize, stream = stream, fileSize = channel.size()) {
          handler.uploadedResult = it
        }
      }
    }
-    finally {
-      unmapBuffer(fileBuffer)
+    else {
+      val fileBuffer = FileChannel.open(file, READ_OPERATION).use { channel ->
+        channel.map(FileChannel.MapMode.READ_ONLY, 0, channel.size())
+      }
+      try {
+        val fileSize = fileBuffer.remaining().toLong()
+
+        zstdCompressContextPool.withZstd(fileSize) { zstd ->
+          compressAndUpload(
+            fileBuffer = fileBuffer,
+            sourceBlockSize = sourceBlockSize,
+            zstd = zstd,
+            stream = stream,
+            fileSize = fileSize,
+          ) {
+            handler.uploadedResult = it
+          }
+        }
+      }
+      finally {
+        unmapBuffer(fileBuffer)
+      }
    }

    // 1. writer must send the last data frame with endStream=true
@@ -76,7 +95,7 @@ private class WebDavPutStatusChecker(private val result: CompletableDeferred<Upl
      result.complete(uploadedResult!!)
    }
    else {
-      result.completeExceptionally(IllegalStateException("Unexpected response status: $status"))
+      result.completeExceptionally(UnexpectedHttpStatus(status))
    }
  }
 }
@@ -118,6 +137,44 @@ private suspend fun compressAndUpload(
    }
    writeFuture.joinCancellable()

+    if (endStream) {
+      return
+    }
+  }
+}
+
+private suspend fun uploadUncompressed(
+  fileChannel: FileChannel,
+  sourceBlockSize: Int,
+  stream: Http2StreamChannel,
+  fileSize: Long,
+  uploadResultConsumer: (UploadResult) -> Unit,
+) {
+  var position = 0L
+  var uploadedSize = 0L
+  while (true) {
+    val chunkSize = min(fileSize - position, sourceBlockSize.toLong()).toInt()
+    val targetNettyBuffer = stream.alloc().directBuffer(chunkSize)
+
+    var written = 0
+    while (written != chunkSize) {
+      val n = targetNettyBuffer.writeBytes(fileChannel, position, chunkSize)
+      if (n == -1) {
+        throw RuntimeException("unexpected EOF")
+      }
+      written += n
+    }
+
+    position += chunkSize
+    uploadedSize += chunkSize
+
+    val endStream = position >= fileSize
+    val writeFuture = stream.writeAndFlush(DefaultHttp2DataFrame(targetNettyBuffer, endStream))
+    if (endStream) {
+      uploadResultConsumer(UploadResult(uploadedSize = uploadedSize, fileSize = fileSize))
+    }
+    writeFuture.joinCancellable()
+
    if (endStream) {
      return
    }
@@ -8,8 +8,8 @@ import io.opentelemetry.api.trace.Span
 import kotlinx.coroutines.Dispatchers
 import org.jetbrains.intellij.build.CompilationContext
 import org.jetbrains.intellij.build.impl.logging.jps.withJpsLogging
+import org.jetbrains.intellij.build.telemetry.TraceManager
 import org.jetbrains.intellij.build.telemetry.TraceManager.spanBuilder
-import org.jetbrains.intellij.build.telemetry.block
 import org.jetbrains.intellij.build.telemetry.use
 import org.jetbrains.jps.api.CanceledStatus
 import org.jetbrains.jps.api.CmdlineRemoteProto.Message.ControllerMessage.ParametersMessage.TargetTypeBuildScope
@@ -279,7 +279,8 @@ internal class JpsCompilationRunner(private val context: CompilationContext) {
        TargetTypeBuildScope.newBuilder()
          .setTypeId(RuntimeModuleRepositoryBuildConstants.TARGET_TYPE_ID)
          .setForceBuild(false)
-          .setAllTargets(true).build()
+          .setAllTargets(true)
+          .build()
      )
    }

@@ -299,7 +300,7 @@ internal class JpsCompilationRunner(private val context: CompilationContext) {
      .setAttribute("modules", moduleSet.joinToString(separator = ", "))
      .setAttribute("incremental", context.options.incrementalCompilation)
      .setAttribute("cacheDir", compilationData.dataStorageRoot.toString())
-      .block(Dispatchers.IO) { span ->
+      .use(Dispatchers.IO) { span ->
        withJpsLogging(context, span) { messageHandler ->
          Standalone.runBuild(
            { context.projectModel },
@@ -311,18 +312,27 @@ internal class JpsCompilationRunner(private val context: CompilationContext) {
            canceledStatus,
          )

-          if (!messageHandler.errorMessagesByCompiler.isEmpty) {
-            for ((key, value) in messageHandler.errorMessagesByCompiler.entrySet()) {
-              @Suppress("UNCHECKED_CAST")
-              context.messages.compilationErrors(key, value as List<String>)
+          if (!messageHandler.errorMessagesByCompiler.isEmpty()) {
+            for ((key, value) in messageHandler.errorMessagesByCompiler) {
+              span.addEvent(
+                "compilation error",
+                Attributes.of(
+                  AttributeKey.stringKey("compiler"), key,
+                  AttributeKey.stringArrayKey("errors"), value,
+                )
+              )
+
+              context.messages.compilationErrors(key, value)
            }
+            span.recordException(RuntimeException("Compilation failed"))
+            TraceManager.scheduleExportPendingSpans()
            throw RuntimeException("Compilation failed")
          }
          else if (!compilationData.statisticsReported && context.options.compilationLogEnabled) {
            messageHandler.printPerModuleCompilationStatistics(compilationStart)
            context.messages.reportStatisticValue(
              "Compilation time, ms",
-              TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - compilationStart).toString()
+              TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - compilationStart).toString(),
            )
            compilationData.statisticsReported = true
          }
@@ -37,9 +37,10 @@ import java.util.regex.Pattern
 import kotlin.io.path.absolutePathString
 import kotlin.io.path.readLines

+private const val NO_TESTS_ERROR = 42
+
 internal class TestingTasksImpl(context: CompilationContext, private val options: TestingOptions) : TestingTasks {
  private val context: CompilationContext = if (options.useArchivedCompiledClasses) ArchivedCompilationContext(context) else context
-  private val NO_TESTS_ERROR = 42

  private fun loadRunConfigurations(name: String): List<JUnitRunConfigurationProperties> {
    return try {
@@ -259,13 +260,13 @@ internal class TestingTasksImpl(context: CompilationContext, private val options

    val testDiscovery = "intellij-test-discovery"
    val library = context.projectModel.project.libraryCollection.findLibrary(testDiscovery)
-                  ?: throw RuntimeException("Can't find the ${testDiscovery} library, but test discovery capturing enabled.")
+                  ?: throw RuntimeException("Can't find the $testDiscovery library, but test discovery capturing enabled.")

    val agentJar = library.getPaths(JpsOrderRootType.COMPILED)
                     .firstOrNull {
                       val name = it.fileName.toString()
                       name.startsWith("intellij-test-discovery") && name.endsWith(".jar")
-                     } ?: throw RuntimeException("Can't find the agent in ${testDiscovery} library, but test discovery capturing enabled.")
+                     } ?: throw RuntimeException("Can't find the agent in $testDiscovery library, but test discovery capturing enabled.")

    additionalJvmOptions += "-javaagent:${agentJar}"
    val excludeRoots = context.projectModel.global.libraryCollection.getLibraries(JpsJavaSdkType.INSTANCE)
@@ -292,7 +293,7 @@ internal class TestingTasksImpl(context: CompilationContext, private val options
    val testConfigurationType = System.getProperty("teamcity.remote-debug.type")
    if (testConfigurationType != "junit") {
      context.messages.error(
-        "Remote debugging is supported for junit run configurations only, but 'teamcity.remote-debug.type' is ${testConfigurationType}")
+        "Remote debugging is supported for junit run configurations only, but 'teamcity.remote-debug.type' is $testConfigurationType")
    }
    val testObject = System.getProperty("teamcity.remote-debug.junit.type")
    val junitClass = System.getProperty("teamcity.remote-debug.junit.class")
@@ -300,7 +301,7 @@ internal class TestingTasksImpl(context: CompilationContext, private val options
      val message = "Remote debugging supports debugging all test methods in a class for now, debugging isn't supported for '${testObject}'"
      if (testObject == "method") {
        context.messages.warning(message)
-        context.messages.warning("Launching all test methods in the class ${junitClass}")
+        context.messages.warning("Launching all test methods in the class $junitClass")
      }
      else {
        context.messages.error(message)
@@ -386,7 +387,7 @@ internal class TestingTasksImpl(context: CompilationContext, private val options
    prepareEnvForTestRun(allJvmArgs, systemProperties, bootstrapClasspath, remoteDebugging)
    val messages = context.messages
    if (isRunningInBatchMode) {
-      messages.info("Running tests from ${mainModule} matched by '${options.batchTestIncludes}' pattern.")
+      messages.info("Running tests from $mainModule matched by '${options.batchTestIncludes}' pattern.")
    }
    else {
      messages.info("Starting tests from groups '${testGroups}' from classpath of module '${mainModule}'")
@@ -397,20 +398,20 @@ internal class TestingTasksImpl(context: CompilationContext, private val options
    spanBuilder("test classpath and runtime info").use {
      withContext(Dispatchers.IO) {
        val runtime = getRuntimeExecutablePath().toString()
-        messages.info("Runtime: ${runtime}")
+        messages.info("Runtime: $runtime")
        runProcess(args = listOf(runtime, "-version"), inheritOut = true, inheritErrToOut = true)
      }

-      messages.info("Runtime options: ${allJvmArgs}")
-      messages.info("System properties: ${systemProperties}")
-      messages.info("Bootstrap classpath: ${bootstrapClasspath}")
-      messages.info("Tests classpath: ${testClasspath}")
+      messages.info("Runtime options: $allJvmArgs")
+      messages.info("System properties: $systemProperties")
+      messages.info("Bootstrap classpath: $bootstrapClasspath")
+      messages.info("Tests classpath: $testClasspath")
      modulePath?.let { mp ->
        @Suppress("SpellCheckingInspection")
        messages.info("Tests modulepath: $mp")
      }
      if (!envVariables.isEmpty()) {
-        messages.info("Environment variables: ${envVariables}")
+        messages.info("Environment variables: $envVariables")
      }
    }
    runJUnit5Engine(
@@ -448,7 +449,7 @@ internal class TestingTasksImpl(context: CompilationContext, private val options
    if (options.customRuntimePath != null) {
      runtimeDir = Path.of(checkNotNull(options.customRuntimePath))
      check(Files.isDirectory(runtimeDir)) {
-        "Custom Jre path from system property '${TestingOptions.TEST_JRE_PROPERTY}' is missing: ${runtimeDir}"
+        "Custom Jre path from system property '${TestingOptions.TEST_JRE_PROPERTY}' is missing: $runtimeDir"
      }
    }
    else {
@@ -459,7 +460,7 @@ internal class TestingTasksImpl(context: CompilationContext, private val options
    if (SystemInfoRt.isMac && Files.notExists(java)) {
      java = runtimeDir.resolve("Contents/Home/bin/java")
    }
-    check(Files.exists(java)) { "java executable is missing: ${java}" }
+    check(Files.exists(java)) { "java executable is missing: $java" }
    return java
  }

@@ -666,7 +667,7 @@ internal class TestingTasksImpl(context: CompilationContext, private val options
    } ?: listOf()

    if (testClasses.isEmpty()) {
-      throw RuntimeException("No tests were found in ${root} with ${filteringPattern}")
+      throw RuntimeException("No tests were found in $root with $filteringPattern")
    }

    return testClasses
@@ -740,12 +741,12 @@ internal class TestingTasksImpl(context: CompilationContext, private val options
        noTestsInAllClasses = noTestsInAllClasses && noTests
      }
      catch (e: Throwable) {
-        throw RuntimeException("Failed to process ${qName}", e)
+        throw RuntimeException("Failed to process $qName", e)
      }
    }

    if (noTestsInAllClasses) {
-      throw RuntimeException("No tests were found in ${mainModule} with ${pattern}")
+      throw RuntimeException("No tests were found in $mainModule with $pattern")
    }
  }

@@ -1041,7 +1042,7 @@ internal class TestingTasksImpl(context: CompilationContext, private val options

    val argFile = CommandLineWrapperUtil.createArgumentFile(args, Charset.defaultCharset())
    val runtime = runBlocking(Dispatchers.IO) { getRuntimeExecutablePath().toString() }
-    context.messages.info("Starting tests on runtime ${runtime}")
+    context.messages.info("Starting tests on runtime $runtime")
    val builder = ProcessBuilder(runtime, "@" + argFile.absolutePath)
    builder.environment().putAll(envVariables)
    if(TeamCityHelper.isUnderTeamCity) {
@@ -1054,7 +1055,7 @@ internal class TestingTasksImpl(context: CompilationContext, private val options
    }
    val exitCode = builder.start().waitFor()
    if (exitCode != 0 && exitCode != NO_TESTS_ERROR) {
-      context.messages.error("Tests failed with exit code ${exitCode}")
+      context.messages.error("Tests failed with exit code $exitCode")
    }
    return exitCode
  }
@@ -1114,7 +1115,7 @@ private fun removeStandardJvmOptions(vmOptions: List<String>): List<String> {
 private suspend fun publishTestDiscovery(messages: BuildMessages, file: String?) {
  val serverUrl = System.getProperty("intellij.test.discovery.url")
  val token = System.getProperty("intellij.test.discovery.token")
-  messages.info("Trying to upload ${file} into ${serverUrl}.")
+  messages.info("Trying to upload $file into ${serverUrl}.")
  val path = file?.let { Path.of(it) }
  if (path != null && Files.exists(path)) {
    if (serverUrl == null) {
@@ -9,8 +9,8 @@ import io.opentelemetry.api.common.Attributes
 import io.opentelemetry.api.trace.Span
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Dispatchers
-import kotlinx.coroutines.coroutineScope
 import kotlinx.coroutines.launch
+import kotlinx.coroutines.withContext
 import kotlinx.serialization.SerialName
 import kotlinx.serialization.Serializable
 import kotlinx.serialization.encodeToString
@@ -19,21 +19,16 @@ import org.jetbrains.annotations.VisibleForTesting
 import org.jetbrains.intellij.build.BuildMessages
 import org.jetbrains.intellij.build.CompilationContext
 import org.jetbrains.intellij.build.forEachConcurrent
-import org.jetbrains.intellij.build.http2Client.READ_OPERATION
 import org.jetbrains.intellij.build.http2Client.withHttp2ClientConnectionFactory
 import org.jetbrains.intellij.build.io.AddDirEntriesMode
 import org.jetbrains.intellij.build.io.zip
 import org.jetbrains.intellij.build.telemetry.TraceManager.spanBuilder
 import org.jetbrains.intellij.build.telemetry.use
-import java.math.BigInteger
 import java.net.InetSocketAddress
 import java.net.URI
-import java.nio.ByteBuffer
-import java.nio.channels.FileChannel
 import java.nio.file.*
 import java.nio.file.attribute.BasicFileAttributes
 import java.nio.file.attribute.FileTime
-import java.security.MessageDigest
 import java.util.*
 import java.util.concurrent.CancellationException
 import java.util.concurrent.ConcurrentHashMap
@@ -310,7 +305,7 @@ suspend fun fetchAndUnpackCompiledClasses(
      }
      true
    }
-      .forEachConcurrent(Runtime.getRuntime().availableProcessors().coerceAtMost(4)) { item ->
+      .forEachConcurrent(Runtime.getRuntime().availableProcessors().coerceAtLeast(4)) { item ->
        val file = item.file
        when {
          Files.notExists(file) -> toDownload.add(item)
@@ -331,40 +326,9 @@ suspend fun fetchAndUnpackCompiledClasses(
  }

  spanBuilder("cleanup outdated compiled class archives").use {
-    val start = System.nanoTime()
-    var count = 0
-    var bytes = 0L
-    try {
-      val preserve = items.mapTo(HashSet<Path>(items.size)) { it.file }
-      val epoch = FileTime.fromMillis(0)
-      val daysAgo = FileTime.fromMillis(System.currentTimeMillis() - TimeUnit.DAYS.toMillis(4))
-      Files.createDirectories(tempDownloadStorage)
-      // we need to traverse with depth 3 since the first level is [production, test], second level is module name, third is a file
-      Files.walk(tempDownloadStorage, 3, FileVisitOption.FOLLOW_LINKS).use { stream ->
-        stream
-          .filter { !preserve.contains(it) }
-          .forEach { file ->
-            val attr = Files.readAttributes(file, BasicFileAttributes::class.java)
-            if (attr.isRegularFile) {
-              val lastAccessTime = attr.lastAccessTime()
-              if (lastAccessTime > epoch && lastAccessTime < daysAgo) {
-                count++
-                bytes += attr.size()
-                Files.deleteIfExists(file)
-              }
-            }
-          }
-      }
-    }
-    catch (e: Throwable) {
-      Span.current().addEvent("failed to cleanup outdated archives", Attributes.of(AttributeKey.stringKey("error"), e.message ?: ""))
-    }
-
-    reportStatisticValue("compile-parts:verify:time", TimeUnit.NANOSECONDS.toMillis(verifyTime).toString())
-    reportStatisticValue("compile-parts:cleanup:time", TimeUnit.NANOSECONDS.toMillis((System.nanoTime() - start)).toString())
-    reportStatisticValue("compile-parts:removed:bytes", bytes.toString())
-    reportStatisticValue("compile-parts:removed:count", count.toString())
+    cleanupOutdatedCompiledClassArchives(items = items, tempDownloadStorage = tempDownloadStorage, reportStatisticValue = reportStatisticValue)
  }
+  reportStatisticValue("compile-parts:verify:time", TimeUnit.NANOSECONDS.toMillis(verifyTime).toString())

  spanBuilder("fetch compiled classes archives").use {
    val start = System.nanoTime()
@@ -389,12 +353,10 @@ suspend fun fetchAndUnpackCompiledClasses(
  }

  val start = System.nanoTime()
-  spanBuilder("unpack compiled classes archives").use {
-    for (item in toUnpack) {
-      launch {
-        spanBuilder("unpack").setAttribute("name", item.name).use {
-          unpackArchive(item, saveHash)
-        }
+  spanBuilder("unpack compiled classes archives").use(Dispatchers.IO) {
+    toUnpack.forEachConcurrent(Runtime.getRuntime().availableProcessors().coerceAtLeast(4)) { item ->
+      spanBuilder("unpack").setAttribute("name", item.name).use {
+        unpackCompilationPartArchive(item, saveHash)
      }
    }
  }
@@ -403,6 +365,45 @@ suspend fun fetchAndUnpackCompiledClasses(
  reportStatisticValue("compile-parts:unpack:time", TimeUnit.NANOSECONDS.toMillis((System.nanoTime() - start)).toString())
 }

+private fun cleanupOutdatedCompiledClassArchives(
+  items: List<FetchAndUnpackItem>,
+  tempDownloadStorage: Path,
+  reportStatisticValue: (key: String, value: String) -> Unit,
+) {
+  val start = System.nanoTime()
+  var count = 0
+  var bytes = 0L
+  try {
+    val preserve = items.mapTo(HashSet<Path>(items.size)) { it.file }
+    val epoch = FileTime.fromMillis(0)
+    val daysAgo = FileTime.fromMillis(System.currentTimeMillis() - TimeUnit.DAYS.toMillis(4))
+    Files.createDirectories(tempDownloadStorage)
+    // we need to traverse with depth 3 since the first level is [production, test], second level is module name, third is a file
+    Files.walk(tempDownloadStorage, 3, FileVisitOption.FOLLOW_LINKS).use { stream ->
+      stream
+        .filter { !preserve.contains(it) }
+        .forEach { file ->
+          val attr = Files.readAttributes(file, BasicFileAttributes::class.java)
+          if (attr.isRegularFile) {
+            val lastAccessTime = attr.lastAccessTime()
+            if (lastAccessTime > epoch && lastAccessTime < daysAgo) {
+              count++
+              bytes += attr.size()
+              Files.deleteIfExists(file)
+            }
+          }
+        }
+    }
+  }
+  catch (e: Throwable) {
+    Span.current().addEvent("failed to cleanup outdated archives", Attributes.of(AttributeKey.stringKey("error"), e.message ?: ""))
+  }
+
+  reportStatisticValue("compile-parts:cleanup:time", TimeUnit.NANOSECONDS.toMillis((System.nanoTime() - start)).toString())
+  reportStatisticValue("compile-parts:removed:bytes", bytes.toString())
+  reportStatisticValue("compile-parts:removed:count", count.toString())
+}
+
 private suspend fun checkPreviouslyUnpackedDirectories(
  items: List<FetchAndUnpackItem>,
  span: Span,
@@ -415,54 +416,52 @@ private suspend fun checkPreviouslyUnpackedDirectories(
  }

  val start = System.nanoTime()
-  coroutineScope {
-    launch(Dispatchers.IO) {
+  withContext(Dispatchers.IO) {
+    launch {
      spanBuilder("remove stalled directories not present in metadata").setAttribute(AttributeKey.stringArrayKey("keys"), java.util.List.copyOf(metadata.files.keys)).use {
        removeStalledDirs(metadata, classOutput)
      }
    }

-    for (item in items) {
-      launch {
-        val out = item.output
-        if (Files.notExists(out)) {
-          span.addEvent("output directory doesn't exist", Attributes.of(AttributeKey.stringKey("name"), item.name, AttributeKey.stringKey("outDir"), out.toString()))
-          return@launch
-        }
+    items.forEachConcurrent { item ->
+      val out = item.output
+      if (Files.notExists(out)) {
+        span.addEvent("output directory doesn't exist", Attributes.of(AttributeKey.stringKey("name"), item.name, AttributeKey.stringKey("outDir"), out.toString()))
+        return@forEachConcurrent
+      }

-        val hashFile = out.resolve(".hash")
-        if (!Files.isRegularFile(hashFile)) {
-          span.addEvent("no .hash file in output directory", Attributes.of(AttributeKey.stringKey("name"), item.name))
-          out.deleteRecursively()
-          return@launch
-        }
+      val hashFile = out.resolve(".hash")
+      if (!Files.isRegularFile(hashFile)) {
+        span.addEvent("no .hash file in output directory", Attributes.of(AttributeKey.stringKey("name"), item.name))
+        out.deleteRecursively()
+        return@forEachConcurrent
+      }

-        try {
-          val actual = Files.readString(hashFile)
-          if (actual == item.hash) {
-            upToDate.add(item.name)
-          }
-          else {
-            span.addEvent(
-              "output directory hash mismatch",
-              Attributes.of(
-                AttributeKey.stringKey("name"), item.name,
-                AttributeKey.stringKey("expected"), item.hash,
-                AttributeKey.stringKey("actual"), actual,
-              )
+      try {
+        val actual = Files.readString(hashFile)
+        if (actual == item.hash) {
+          upToDate.add(item.name)
+        }
+        else {
+          span.addEvent(
+            "output directory hash mismatch",
+            Attributes.of(
+              AttributeKey.stringKey("name"), item.name,
+              AttributeKey.stringKey("expected"), item.hash,
+              AttributeKey.stringKey("actual"), actual,
            )
-            out.deleteRecursively()
-          }
-        }
-        catch (e: CancellationException) {
-          throw e
-        }
-        catch (e: Throwable) {
-          span.addEvent("output directory hash calculation failed", Attributes.of(AttributeKey.stringKey("name"), item.name))
-          span.recordException(e, Attributes.of(AttributeKey.stringKey("name"), item.name))
+          )
          out.deleteRecursively()
        }
      }
+      catch (e: CancellationException) {
+        throw e
+      }
+      catch (e: Throwable) {
+        span.addEvent("output directory hash calculation failed", Attributes.of(AttributeKey.stringKey("name"), item.name))
+        span.recordException(e, Attributes.of(AttributeKey.stringKey("name"), item.name))
+        out.deleteRecursively()
+      }
    }
  }
  return System.nanoTime() - start
@@ -505,34 +504,6 @@ private fun CoroutineScope.removeStalledDirs(
  }
 }

-private val sharedDigest = MessageDigest.getInstance("SHA-256", java.security.Security.getProvider("SUN"))
-internal fun sha256() = sharedDigest.clone() as MessageDigest
-
-internal fun computeHash(file: Path): String {
-  val messageDigest = sha256()
-  FileChannel.open(file, READ_OPERATION).use { channel ->
-    val fileSize = channel.size()
-    // java message digest doesn't support native buffer (copies to a heap byte array in any case)
-    val bufferSize = 256 * 1024
-    val buffer = ByteBuffer.allocate(bufferSize)
-    var offset = 0L
-    var readBytes: Int
-    while (offset < fileSize) {
-      buffer.clear()
-      readBytes = channel.read(buffer, offset)
-      if (readBytes <= 0) {
-        break
-      }
-
-      messageDigest.update(buffer.array(), 0, readBytes)
-      offset += readBytes
-    }
-  }
-  return digestToString(messageDigest)
-}
-
-internal fun digestToString(digest: MessageDigest): String = BigInteger(1, digest.digest()).toString(36) + "z"
-
 internal data class PackAndUploadItem(
  @JvmField val output: Path,
  @JvmField val name: String,
@@ -9,7 +9,6 @@ import io.opentelemetry.api.common.Attributes
 import io.opentelemetry.api.trace.Span
 import kotlinx.coroutines.TimeoutCancellationException
 import kotlinx.coroutines.isActive
-import kotlinx.coroutines.launch
 import kotlinx.coroutines.withTimeout
 import org.jetbrains.intellij.build.BuildOptions
 import org.jetbrains.intellij.build.CompilationContext
@@ -22,8 +21,6 @@ import org.jetbrains.intellij.build.telemetry.use
 import org.jetbrains.jps.api.CanceledStatus
 import org.jetbrains.jps.incremental.storage.ProjectStamps
 import java.nio.file.Path
-import kotlin.time.Duration
-import kotlin.time.Duration.Companion.minutes

 internal fun checkCompilationOptions(context: CompilationContext) {
  val options = context.options
@@ -189,18 +186,20 @@ internal suspend fun doCompile(
  check(isCompilationRequired(context.options)) {
    "Unexpected compilation request, unable to proceed"
  }
-  context.messages.progress("Compiling project")
  context.compilationData.statisticsReported = false
  val runner = JpsCompilationRunner(context)
  try {
    val (status, isIncrementalCompilation) = when {
      context.options.forceRebuild -> "forced rebuild" to false
      availableCommitDepth >= 0 -> portableJpsCacheUsageStatus(availableCommitDepth) to true
-      isIncrementalCompilationDataAvailable(context) -> "compiled using local cache" to true
+      isIncrementalCompilationDataAvailable(context) -> "compile using local cache" to true
      else -> "clean build" to false
    }
    context.options.incrementalCompilation = isIncrementalCompilation
-    if (!isIncrementalCompilation) {
+    if (isIncrementalCompilation) {
+      Span.current().addEvent("status: $status")
+    }
+    else {
      Span.current().addEvent(
        "no compiled classes can be reused",
        Attributes.of(
@@ -209,20 +208,22 @@ internal suspend fun doCompile(
        )
      )
    }
-    block(status) {
-      if (isIncrementalCompilation) {
-        // workaround for KT-55695
-        compileWithTimeout(
-          compilationRunner = runner,
+
+    val incrementalCompilationTimeout = context.options.incrementalCompilationTimeout
+    if (isIncrementalCompilation && incrementalCompilationTimeout != null) {
+      // workaround for KT-55695
+      withTimeout(incrementalCompilationTimeout) {
+        compile(
+          jpsCompilationRunner = runner,
          context = context,
          moduleNames = moduleNames,
          includingTestsInModules = includingTestsInModules,
-          timeout = context.options.incrementalCompilationTimeout.minutes,
+          canceledStatus = CanceledStatus { !isActive },
        )
      }
-      else {
-        compile(jpsCompilationRunner = runner, context = context, moduleNames = moduleNames, includingTestsInModules = includingTestsInModules)
-      }
+    }
+    else {
+      compile(jpsCompilationRunner = runner, context = context, moduleNames = moduleNames, includingTestsInModules = includingTestsInModules)
    }
    context.messages.buildStatus(status)
  }
@@ -246,26 +247,6 @@ private suspend fun unpackCompiledClasses(classOutput: Path, context: Compilatio
  }
 }

-private suspend fun compileWithTimeout(
-  compilationRunner: JpsCompilationRunner,
-  context: CompilationContext,
-  moduleNames: Collection<String>?,
-  includingTestsInModules: List<String>?,
-  timeout: Duration,
-) {
-  withTimeout(timeout) {
-    launch {
-      compile(
-        jpsCompilationRunner = compilationRunner,
-        context = context,
-        moduleNames = moduleNames,
-        includingTestsInModules = includingTestsInModules,
-        canceledStatus = CanceledStatus { !isActive },
-      )
-    }
-  }
-}
-
 private suspend fun retryCompilation(
  context: CompilationContext,
  runner: JpsCompilationRunner,
@@ -1,6 +1,7 @@
 // Copyright 2000-2024 JetBrains s.r.o. and contributors. Use of this source code is governed by the Apache 2.0 license.
 package org.jetbrains.intellij.build.impl.compilation

+import io.netty.util.AsciiString
 import io.opentelemetry.api.trace.Span
 import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.runBlocking
@@ -14,12 +15,13 @@ import org.jetbrains.intellij.build.telemetry.TraceManager.spanBuilder
 import org.jetbrains.intellij.build.telemetry.block
 import org.jetbrains.intellij.build.telemetry.use
 import org.jetbrains.jps.incremental.storage.ProjectStamps
+import java.net.URI
 import java.nio.file.Path
 import java.util.*
 import java.util.concurrent.CancellationException

 internal val IS_PORTABLE_COMPILATION_CACHE_ENABLED: Boolean
-  get() = ProjectStamps.PORTABLE_CACHES && IS_CONFIGURED
+  get() = ProjectStamps.PORTABLE_CACHES && IS_JPS_CACHE_URL_CONFIGURED

 private var isAlreadyUpdated = false

@@ -29,7 +31,7 @@ internal object TestJpsCompilationCacheDownload {
    System.setProperty("jps.cache.test", "true")
    System.setProperty("org.jetbrains.jps.portable.caches", "true")
    if (System.getProperty(URL_PROPERTY) == null) {
-      System.setProperty(URL_PROPERTY, "https://127.0.0.1:1900")
+      System.setProperty(URL_PROPERTY, "https://127.0.0.1:1900/cache/jps")
    }

    val projectHome = ULTIMATE_HOME
@@ -42,34 +44,55 @@ internal object TestJpsCompilationCacheDownload {
        useCompiledClassesFromProjectOutput = true,
      ),
    )
-    downloadCacheAndCompileProject(forceDownload = false, context = context)
+    downloadCacheAndCompileProject(forceDownload = false, gitUrl = computeRemoteGitUrl(), context = context)
  }
 }

 internal suspend fun downloadJpsCacheAndCompileProject(context: CompilationContext) {
-  downloadCacheAndCompileProject(forceDownload = System.getProperty(FORCE_DOWNLOAD_PROPERTY).toBoolean(), context = context)
+  downloadCacheAndCompileProject(forceDownload = System.getProperty(FORCE_DOWNLOAD_PROPERTY).toBoolean(), gitUrl = computeRemoteGitUrl(), context = context)
 }

 /**
- * Upload local [PortableCompilationCache] to [PortableJpsCacheRemoteCacheConfig]
+ * Upload local [PortableCompilationCache].
 */
 suspend fun uploadPortableCompilationCache(context: CompilationContext) {
-  createPortableCompilationCacheUploader(context).upload(context.messages)
+  uploadJpsCache(
+    forcedUpload = context.options.forceRebuild,
+    commitHash = getCommitCache(),
+    s3Dir = getS3Dir(),
+    authHeader = getAuthHeader(),
+    uploadUrl = getJpsCacheUploadUrl(),
+    context = context,
+  )
 }

 /**
- * Publish already uploaded [PortableCompilationCache] to [PortableJpsCacheRemoteCacheConfig]
+ * Publish already uploaded [PortableCompilationCache].
 */
-suspend fun publishPortableCompilationCache(context: CompilationContext) {
-  createPortableCompilationCacheUploader(context).updateCommitHistory()
+suspend fun publishPortableCompilationCache(context: CompilationContext, overrideCommits: Set<String>? = null) {
+  updateJpsCacheCommitHistory(
+    overrideCommits = overrideCommits,
+    remoteGitUrl = computeRemoteGitUrl(),
+    commitHash = getCommitCache(),
+    uploadUrl = getJpsCacheUploadUrl(),
+    authHeader = getAuthHeader(),
+    context = context,
+    s3Dir = getS3Dir(),
+  )
+}
+
+private fun getCommitCache() = require(COMMIT_HASH_PROPERTY, "Repository commit")
+
+private fun getS3Dir(): Path? {
+  return System.getProperty(AWS_SYNC_FOLDER_PROPERTY)?.let<String, Path> { Path.of(it) }
 }

 /**
- * Publish already uploaded [PortableCompilationCache] to [PortableJpsCacheRemoteCacheConfig] overriding existing [CommitsHistory].
+ * Publish already uploaded [PortableCompilationCache] overriding existing [CommitsHistory].
 * Used in force rebuild and cleanup.
 */
 suspend fun publishUploadedJpsCacheWithCommitHistoryOverride(forceRebuiltCommits: Set<String>, context: CompilationContext) {
-  createPortableCompilationCacheUploader(context).updateCommitHistory(overrideCommits = forceRebuiltCommits)
+  publishPortableCompilationCache(context = context, overrideCommits = forceRebuiltCommits)
 }

 /**
@@ -84,7 +107,7 @@ suspend fun publishUploadedJpsCacheWithCommitHistoryOverride(forceRebuiltCommits
 * For more details see [org.jetbrains.jps.backwardRefs.JavaBackwardReferenceIndexWriter.initialize]
 */
@Suppress("KDocUnresolvedReference")
-private suspend fun downloadCacheAndCompileProject(forceDownload: Boolean, context: CompilationContext) {
+private suspend fun downloadCacheAndCompileProject(forceDownload: Boolean, gitUrl: String, context: CompilationContext) {
  val forceRebuild = context.options.forceRebuild
  spanBuilder("download JPS cache and compile")
    .setAttribute("forceRebuild", forceRebuild)
@@ -103,16 +126,11 @@ private suspend fun downloadCacheAndCompileProject(forceDownload: Boolean, conte
        cleanOutput(context = context, keepCompilationState = false)
      }

-      val downloader = PortableCompilationCacheDownloader(
-        context = context,
-        git = Git(context.paths.projectHome),
-        remoteCache = PortableJpsCacheRemoteCacheConfig(),
-        gitUrl = computeRemoteGitUrl(),
-      )
+      val downloader = PortableCompilationCacheDownloader(context = context, git = Git(context.paths.projectHome))

      val portableCompilationCache = PortableCompilationCache(forceDownload = forceDownload)
      val availableCommitDepth = if (!forceRebuild && (forceDownload || !isIncrementalCompilationDataAvailable(context))) {
-        portableCompilationCache.downloadCache(downloader, context)
+        portableCompilationCache.downloadCache(downloader = downloader, gitUrl = computeRemoteGitUrl(), context = context)
      }
      else {
        -1
@@ -121,18 +139,21 @@ private suspend fun downloadCacheAndCompileProject(forceDownload: Boolean, conte
      context.options.incrementalCompilation = !forceRebuild
      // compilation is executed unconditionally here even if the exact commit cache is downloaded
      // to have an additional validation step and not to ignore a local changes, for example, in TeamCity Remote Run
-      doCompile(
-        availableCommitDepth = availableCommitDepth,
-        context = context,
-        handleCompilationFailureBeforeRetry = { successMessage ->
-          portableCompilationCache.handleCompilationFailureBeforeRetry(
-            successMessage = successMessage,
-            portableCompilationCacheDownloader = downloader,
-            forceDownload = portableCompilationCache.forceDownload,
-            context = context,
-          )
-        },
-      )
+      if (!context.options.useCompiledClassesFromProjectOutput) {
+        doCompile(
+          availableCommitDepth = availableCommitDepth,
+          context = context,
+          handleCompilationFailureBeforeRetry = { successMessage ->
+            portableCompilationCache.handleCompilationFailureBeforeRetry(
+              successMessage = successMessage,
+              portableCompilationCacheDownloader = downloader,
+              forceDownload = portableCompilationCache.forceDownload,
+              gitUrl = gitUrl,
+              context = context,
+            )
+          },
+        )
+      }
      isAlreadyUpdated = true
      context.options.incrementalCompilation = true
    }
@@ -150,6 +171,7 @@ private class PortableCompilationCache(forceDownload: Boolean) {
    portableCompilationCacheDownloader: PortableCompilationCacheDownloader,
    context: CompilationContext,
    forceDownload: Boolean,
+    gitUrl: String,
  ): String {
    when {
      forceDownload -> {
@@ -162,7 +184,7 @@ private class PortableCompilationCache(forceDownload: Boolean) {
        // If download isn't forced, then locally available cache will be used which may suffer from those issues.
        // Hence, compilation failure. Replacing local cache with remote one may help.
        Span.current().addEvent("Incremental compilation using locally available caches failed. Re-trying using Remote Cache.")
-        val availableCommitDepth = downloadCache(portableCompilationCacheDownloader, context)
+        val availableCommitDepth = downloadCache(portableCompilationCacheDownloader, gitUrl = gitUrl, context)
        if (availableCommitDepth >= 0) {
          return portableJpsCacheUsageStatus(availableCommitDepth)
        }
@@ -171,10 +193,10 @@ private class PortableCompilationCache(forceDownload: Boolean) {
    return successMessage
  }

-  suspend fun downloadCache(downloader: PortableCompilationCacheDownloader, context: CompilationContext): Int {
+  suspend fun downloadCache(downloader: PortableCompilationCacheDownloader, gitUrl: String, context: CompilationContext): Int {
    return spanBuilder("downloading Portable Compilation Cache").use { span ->
      try {
-        downloader.download()
+        downloader.download(cacheUrl = require(URL_PROPERTY, "Remote Cache url"), gitUrl = gitUrl, authHeader = getAuthHeader())
      }
      catch (e: CancellationException) {
        throw e
@@ -211,10 +233,7 @@ private const val UPLOAD_URL_PROPERTY = "intellij.jps.remote.cache.upload.url"
 */
 private const val URL_PROPERTY = "intellij.jps.remote.cache.url"

-/**
- * If true then [PortableJpsCacheRemoteCacheConfig] is configured to be used
- */
-private val IS_CONFIGURED = !System.getProperty(URL_PROPERTY).isNullOrBlank()
+private val IS_JPS_CACHE_URL_CONFIGURED = !System.getProperty(URL_PROPERTY).isNullOrBlank()

 /**
 * IntelliJ repository git remote url
@@ -256,32 +275,16 @@ internal class CompilationOutput(
  @JvmField val path: Path,
 )

-/**
- * Server which stores [PortableCompilationCache]
- */
-internal class PortableJpsCacheRemoteCacheConfig {
-  val url by lazy { require(URL_PROPERTY, "Remote Cache url") }
-  val uploadUrl by lazy { require(UPLOAD_URL_PROPERTY, "Remote Cache upload url") }
-  val authHeader by lazy {
-    val username = System.getProperty("jps.auth.spaceUsername")
-    val password = System.getProperty("jps.auth.spacePassword")
-    when {
-      password == null -> ""
-      username == null -> "Bearer $password"
-      else -> "Basic " + Base64.getEncoder().encodeToString("$username:$password".toByteArray())
-    }
-  }
-}
+private fun getJpsCacheUploadUrl(): URI = URI(require(UPLOAD_URL_PROPERTY, "Remote Cache upload url"))

-internal fun createPortableCompilationCacheUploader(context: CompilationContext): PortableCompilationCacheUploader {
-  return PortableCompilationCacheUploader(
-    context = context,
-    remoteCache = PortableJpsCacheRemoteCacheConfig(),
-    remoteGitUrl = computeRemoteGitUrl(),
-    commitHash = require(COMMIT_HASH_PROPERTY, "Repository commit"),
-    s3Folder = Path.of(require(AWS_SYNC_FOLDER_PROPERTY, "AWS S3 sync folder")),
-    forcedUpload = context.options.forceRebuild,
-  )
+private fun getAuthHeader(): CharSequence {
+  val username = System.getProperty("jps.auth.spaceUsername")
+  val password = System.getProperty("jps.auth.spacePassword")
+  return when {
+    password == null -> AsciiString.EMPTY_STRING
+    username == null -> AsciiString.of("Bearer $password")
+    else -> AsciiString.of("Basic " + Base64.getEncoder().encodeToString("$username:$password".toByteArray()))
+  }
 }

 private fun computeRemoteGitUrl(): String {
@@ -3,22 +3,23 @@

 package org.jetbrains.intellij.build.impl.compilation

-import com.intellij.util.io.Decompressor
+import com.google.gson.stream.JsonReader
 import io.opentelemetry.api.common.AttributeKey
 import io.opentelemetry.api.common.Attributes
 import io.opentelemetry.api.trace.Span
-import kotlinx.coroutines.*
-import okhttp3.Request
-import okio.sink
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.coroutineScope
+import kotlinx.coroutines.launch
 import org.jetbrains.intellij.build.CompilationContext
 import org.jetbrains.intellij.build.forEachConcurrent
+import org.jetbrains.intellij.build.http2Client.*
 import org.jetbrains.intellij.build.impl.compilation.cache.CommitsHistory
 import org.jetbrains.intellij.build.impl.compilation.cache.getAllCompilationOutputs
-import org.jetbrains.intellij.build.impl.compilation.cache.parseSourcesStateFile
 import org.jetbrains.intellij.build.retryWithExponentialBackOff
 import org.jetbrains.intellij.build.telemetry.TraceManager.spanBuilder
 import org.jetbrains.intellij.build.telemetry.use
-import java.io.IOException
+import org.jetbrains.jps.incremental.storage.BuildTargetSourcesState
+import java.net.URI
 import java.nio.file.Files
 import java.nio.file.NoSuchFileException
 import java.nio.file.Path
@@ -28,83 +29,53 @@ import java.util.concurrent.atomic.LongAdder

 private const val COMMITS_COUNT = 1_000

-internal class PortableCompilationCacheDownloader(
-  private val context: CompilationContext,
-  private val git: Git,
-  private val remoteCache: PortableJpsCacheRemoteCacheConfig,
-  private val gitUrl: String,
-) {
-  private val remoteCacheUrl = remoteCache.url.trimEnd('/')
-
+internal class PortableCompilationCacheDownloader(private val context: CompilationContext, private val git: Git) {
  private val lastCommits by lazy { git.log(COMMITS_COUNT) }

-  private suspend fun downloadString(url: String): String = retryWithExponentialBackOff {
-    if (url.isS3()) {
-      awsS3Cli("cp", url, "-")
-    }
-    else {
-      httpClient.get(url, remoteCache.authHeader) { it.body.string() }
-    }
-  }
-
-  private suspend fun downloadToFile(url: String, file: Path, spanName: String, notFound: LongAdder) {
-    spanBuilder(spanName).setAttribute("url", url).setAttribute("path", file.toString()).use { span ->
-      Files.createDirectories(file.parent)
-      retryWithExponentialBackOff {
-        if (url.isS3()) {
-          awsS3Cli("cp", url, file.toString())
+  private suspend fun downloadToFile(urlPath: String, file: Path, spanName: String, notFound: LongAdder, connection: Http2ClientConnection?): Long {
+    return spanBuilder(spanName).setAttribute("urlPath", urlPath).setAttribute("path", file.toString()).use { span ->
+      if (connection == null) {
+        Files.createDirectories(file.parent)
+        require(urlPath.isS3())
+        retryWithExponentialBackOff {
+          awsS3Cli("cp", urlPath, file.toString())
        }
-        else {
-          httpClient.newCall(Request.Builder().url(url).header("Authorization", remoteCache.authHeader).build())
-            .executeAsync()
-            .use { response ->
-              when {
-                response.isSuccessful -> {
-                  file.sink().use {
-                    response.body.source().readAll(it)
-                  }
-                }
-                response.code == 404 -> {
-                  span.addEvent("resource not found")
-                  notFound.increment()
-                }
-                else -> throw IOException("Unexpected code $response")
-              }
-            }
+        return@use try {
+          Files.size(file)
+        }
+        catch (e: NoSuchFileException) {
+          0
        }
      }
-    }
-  }

-  @OptIn(DelicateCoroutinesApi::class)
-  private val availableCachesKeysLazyTask = GlobalScope.async(Dispatchers.Unconfined, start = CoroutineStart.LAZY) {
-    val commitHistory = "$remoteCacheUrl/${CommitsHistory.JSON_FILE}"
-    if (isExist(commitHistory)) {
-      val json = downloadString(commitHistory)
-      CommitsHistory(json).commitsForRemote(gitUrl)
-    }
-    else {
-      emptyList()
-    }
-  }
-
-  private suspend fun isExist(path: String): Boolean {
-    return spanBuilder("head").setAttribute("url", remoteCacheUrl).use {
-      retryWithExponentialBackOff {
-        httpClient.head(path, remoteCache.authHeader) == 200
+      val sizeOnDisk = connection.download(path = urlPath, file = file)
+      if (sizeOnDisk == -1L) {
+        span.addEvent("resource not found")
+        notFound.increment()
      }
+      sizeOnDisk
    }
  }

-  suspend fun download(): Int {
-    val availableCachesKeys = availableCachesKeysLazyTask.await()
+  private suspend fun getAvailableCachesKeysLazyTask(urlPathPrefix: String, gitUrl: String, connection: Http2ClientConnection?): Collection<String> {
+    val commitHistoryUrl = "$urlPathPrefix/${CommitsHistory.JSON_FILE}"
+    require(!commitHistoryUrl.isS3() && connection != null)
+    val json: Map<String, Set<String>> = connection.getJsonOrDefaultIfNotFound(path = commitHistoryUrl, defaultIfNotFound = emptyMap())
+    if (json.isEmpty()) {
+      return emptyList()
+    }
+    return CommitsHistory(json).commitsForRemote(gitUrl)
+  }
+
+  private suspend fun prepareDownload(urlPathPrefix: String, gitUrl: String, connection: Http2ClientConnection?): Pair<String, Int>? {
+    val availableCachesKeys = getAvailableCachesKeysLazyTask(urlPathPrefix = urlPathPrefix, gitUrl = gitUrl, connection = connection)
    val availableCommitDepth = lastCommits.indexOfFirst {
      availableCachesKeys.contains(it)
    }

    if (availableCommitDepth !in 0 until lastCommits.count()) {
-      Span.current().addEvent("Unable to find cache for any of last ${lastCommits.count()} commits.")
-      return -1
+      Span.current().addEvent("unable to find cache for any of last ${lastCommits.count()} commits.")
+      return null
    }

    val lastCachedCommit = lastCommits.get(availableCommitDepth)
@@ -112,30 +83,47 @@ internal class PortableCompilationCacheDownloader(
      "using cache for commit $lastCachedCommit ($availableCommitDepth behind last commit)",
      Attributes.of(
        AttributeKey.longKey("behind last commit"), availableCommitDepth.toLong(),
-        AttributeKey.stringArrayKey("available cache keys"), availableCachesKeysLazyTask.await().toList(),
+        AttributeKey.stringArrayKey("available cache keys"), availableCachesKeys.toList(),
      )
    )
+    return lastCachedCommit to availableCommitDepth
+  }
+
+  suspend fun download(cacheUrl: String, authHeader: CharSequence, gitUrl: String): Int {
    val start = System.nanoTime()
    val totalDownloadedBytes = LongAdder()
    val notFound = LongAdder()
-    var total = -1
-    withContext(Dispatchers.IO) {
-      launch {
-        spanBuilder("get and unpack jps cache").setAttribute("commit", lastCachedCommit).use {
-          downloadAndUnpackJpsCache(commitHash = lastCachedCommit, notFound = notFound, totalBytes = totalDownloadedBytes)
+    var availableCommitDepth = -1
+    val total = if (cacheUrl.isS3()) {
+      val info = prepareDownload(urlPathPrefix = cacheUrl, gitUrl = gitUrl, connection = null) ?: return -1
+      availableCommitDepth = info.second
+      doDownload(
+        urlPathPrefix = cacheUrl,
+        lastCachedCommit = info.first,
+        notFound = notFound,
+        totalDownloadedBytes = totalDownloadedBytes,
+        connection = null,
+      )
+    }
+    else {
+      val serverUri = URI(cacheUrl)
+      withHttp2ClientConnectionFactory(trustAll = serverUri.host == "127.0.0.1") { client ->
+        client.connect(serverUri.host, serverUri.port).withAuth(authHeader).use { connection ->
+          val info = prepareDownload(urlPathPrefix = serverUri.path, gitUrl = gitUrl, connection = connection) ?: return@use -1
+          availableCommitDepth = info.second
+          doDownload(
+            urlPathPrefix = serverUri.path,
+            lastCachedCommit = info.first,
+            notFound = notFound,
+            totalDownloadedBytes = totalDownloadedBytes,
+            connection = connection,
+          )
        }
      }
+    }

-      val sourcesState = parseSourcesStateFile(downloadString("$remoteCacheUrl/metadata/$lastCachedCommit"))
-      val outputs = getAllCompilationOutputs(sourcesState, context.classesOutputDirectory)
-      total = outputs.size
-      spanBuilder("download compilation output parts").setAttribute(AttributeKey.longKey("count"), outputs.size.toLong()).use {
-        outputs.forEachConcurrent(downloadParallelism) { output ->
-          spanBuilder("get and unpack output").setAttribute("part", output.remotePath).use {
-            downloadAndUnpackCompilationOutput(compilationOutput = output, notFound = notFound, totalDownloadedBytes = totalDownloadedBytes)
-          }
-        }
-      }
+    if (availableCommitDepth == -1) {
+      return -1
    }

    reportStatisticValue("jps-cache:download:time", TimeUnit.NANOSECONDS.toMillis((System.nanoTime() - start)).toString())
@@ -146,19 +134,75 @@ internal class PortableCompilationCacheDownloader(
    return availableCommitDepth
  }

-  private suspend fun downloadAndUnpackJpsCache(commitHash: String, notFound: LongAdder, totalBytes: LongAdder) {
+  private suspend fun PortableCompilationCacheDownloader.doDownload(
+    urlPathPrefix: String,
+    lastCachedCommit: String,
+    notFound: LongAdder,
+    totalDownloadedBytes: LongAdder,
+    connection: Http2ClientConnection?,
+  ): Int {
+    return coroutineScope {
+      launch {
+        spanBuilder("get and unpack jps cache").setAttribute("commit", lastCachedCommit).use {
+          downloadAndUnpackJpsCache(
+            urlPathPrefix = urlPathPrefix,
+            commitHash = lastCachedCommit,
+            notFound = notFound,
+            totalBytes = totalDownloadedBytes,
+            connection = connection,
+          )
+        }
+      }
+
+      val metadataUrlPath = "$urlPathPrefix/metadata/$lastCachedCommit"
+      val json = if (connection == null) {
+        require(metadataUrlPath.isS3())
+        retryWithExponentialBackOff {
+          awsS3Cli("cp", metadataUrlPath, "-")
+        }
+      }
+      else {
+        connection.getString(path = metadataUrlPath)
+      }
+
+      val outputs = getAllCompilationOutputs(sourceState = BuildTargetSourcesState.readJson(JsonReader(json.reader())), classOutDir = context.classesOutputDirectory)
+      spanBuilder("download compilation output parts").setAttribute(AttributeKey.longKey("count"), outputs.size.toLong()).use {
+        outputs.forEachConcurrent(downloadParallelism) { output ->
+          spanBuilder("get and unpack output").setAttribute("part", output.remotePath).use {
+            downloadAndUnpackCompilationOutput(
+              urlPathPrefix = urlPathPrefix,
+              compilationOutput = output,
+              notFound = notFound,
+              totalDownloadedBytes = totalDownloadedBytes,
+              connection = connection,
+            )
+          }
+        }
+      }
+      outputs.size
+    }
+  }
+
+  private suspend fun downloadAndUnpackJpsCache(urlPathPrefix: String, commitHash: String, notFound: LongAdder, totalBytes: LongAdder, connection: Http2ClientConnection?) {
    val cacheArchive = Files.createTempFile("cache", ".zip")
    try {
-      downloadToFile(url = "$remoteCacheUrl/caches/$commitHash", file = cacheArchive, spanName = "download jps cache", notFound = notFound)
+      val sizeOnDisk = downloadToFile(
+        urlPath = "$urlPathPrefix/caches/$commitHash",
+        file = cacheArchive,
+        spanName = "download jps cache",
+        notFound = notFound,
+        connection = connection,
+      )

-      totalBytes.add(Files.size(cacheArchive))
+      require(sizeOnDisk > 0)
+      totalBytes.add(sizeOnDisk)
      val cacheDestination = context.compilationData.dataStorageRoot

      spanBuilder("unpack jps cache")
-        .setAttribute("archive", "$cacheArchive")
-        .setAttribute("destination", "$cacheDestination")
-        .use {
-          Decompressor.Zip(cacheArchive).overwrite(true).extract(cacheDestination)
+        .setAttribute("archive", cacheArchive.toString())
+        .setAttribute("destination", cacheDestination.toString())
+        .use(Dispatchers.IO) {
+          unpackArchiveUsingNettyByteBufferPool(archiveFile = cacheArchive, outDir = cacheDestination, isCompressed = true)
        }
    }
    finally {
@@ -166,32 +210,45 @@ internal class PortableCompilationCacheDownloader(
    }
  }

-  private suspend fun downloadAndUnpackCompilationOutput(compilationOutput: CompilationOutput, notFound: LongAdder, totalDownloadedBytes: LongAdder) {
+  private suspend fun downloadAndUnpackCompilationOutput(
+    urlPathPrefix: String,
+    compilationOutput: CompilationOutput,
+    notFound: LongAdder,
+    totalDownloadedBytes: LongAdder,
+    connection: Http2ClientConnection?,
+  ) {
    val tempFile = compilationOutput.path.resolve("tmp-output.zip")
-    downloadToFile(url = "$remoteCacheUrl/${compilationOutput.remotePath}", file = tempFile, spanName = "download output", notFound = notFound)
-    try {
-      totalDownloadedBytes.add(Files.size(tempFile))
-    }
-    catch (e: NoSuchFileException) {
-      // We assume that the error is logged by downloadToFile.
-      // In the future, we will implement stricter behavior.
-      // For now, the JPS cache reports non-existent compilation outputs, and we have to use such a workaround.
+    val urlPath = "$urlPathPrefix/${compilationOutput.remotePath.trimStart('/')}"
+    val sizeOnDisk = downloadToFile(
+      urlPath = urlPath,
+      file = tempFile,
+      spanName = "download output",
+      notFound = notFound,
+      connection = connection,
+    )
+
+    if (sizeOnDisk <= 0) {
      return
    }

+    // We assume that the error is logged by downloadToFile.
+    // In the future, we will implement stricter behavior.
+    // For now, the JPS cache reports non-existent compilation outputs, and we have to use such a workaround.
+    totalDownloadedBytes.add(sizeOnDisk)
+
    try {
      spanBuilder("unpack output")
        .setAttribute("archive", tempFile.toString())
        .setAttribute("destination", compilationOutput.path.toString())
-        .use {
-          Decompressor.Zip(tempFile).overwrite(true).extract(compilationOutput.path)
+        .use(Dispatchers.IO) {
+          unpackArchiveUsingNettyByteBufferPool(archiveFile = tempFile, outDir = compilationOutput.path, isCompressed = true)
        }
    }
    catch (e: CancellationException) {
      throw e
    }
    catch (e: Exception) {
-      throw Exception("Unable to unpack $remoteCacheUrl/${compilationOutput.remotePath} to ${compilationOutput.path}", e)
+      throw Exception("Unable to unpack $urlPath to ${compilationOutput.path}", e)
    }
    finally {
      Files.deleteIfExists(tempFile)
@@ -3,270 +3,279 @@
 package org.jetbrains.intellij.build.impl.compilation

 import com.google.gson.stream.JsonReader
-import com.intellij.util.io.Compressor
+import io.netty.handler.codec.http.HttpResponseStatus
 import io.opentelemetry.api.common.AttributeKey
 import io.opentelemetry.api.common.Attributes
 import io.opentelemetry.api.trace.Span
 import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.launch
 import kotlinx.coroutines.withContext
-import okhttp3.MediaType.Companion.toMediaType
-import okhttp3.Request
-import okhttp3.RequestBody
-import okio.BufferedSink
-import okio.source
-import org.jetbrains.intellij.build.BuildMessages
 import org.jetbrains.intellij.build.CompilationContext
 import org.jetbrains.intellij.build.forEachConcurrent
+import org.jetbrains.intellij.build.http2Client.Http2ClientConnection
+import org.jetbrains.intellij.build.http2Client.getJsonOrDefaultIfNotFound
+import org.jetbrains.intellij.build.http2Client.upload
+import org.jetbrains.intellij.build.http2Client.withHttp2ClientConnectionFactory
 import org.jetbrains.intellij.build.impl.compilation.cache.CommitsHistory
 import org.jetbrains.intellij.build.impl.compilation.cache.getAllCompilationOutputs
 import org.jetbrains.intellij.build.io.copyFile
 import org.jetbrains.intellij.build.io.moveFile
 import org.jetbrains.intellij.build.io.zipWithCompression
-import org.jetbrains.intellij.build.retryWithExponentialBackOff
 import org.jetbrains.intellij.build.telemetry.TraceManager.spanBuilder
 import org.jetbrains.intellij.build.telemetry.use
 import org.jetbrains.jps.incremental.storage.BuildTargetSourcesState
 import org.jetbrains.jps.incremental.storage.ProjectStamps
+import java.net.URI
 import java.nio.file.Files
 import java.nio.file.NoSuchFileException
 import java.nio.file.Path
 import java.util.concurrent.TimeUnit
-import java.util.concurrent.atomic.AtomicInteger
-import java.util.concurrent.atomic.AtomicLong
+import java.util.concurrent.atomic.LongAdder
 import kotlin.io.path.ExperimentalPathApi
 import kotlin.io.path.deleteRecursively
+import kotlin.random.Random

 private const val SOURCES_STATE_FILE_NAME = "target_sources_state.json"
-private val MEDIA_TYPE_BINARY = "application/octet-stream".toMediaType()

-internal class PortableCompilationCacheUploader(
-  private val context: CompilationContext,
-  internal val remoteCache: PortableJpsCacheRemoteCacheConfig,
-  @JvmField val remoteGitUrl: String,
-  private val commitHash: String,
-  private val s3Folder: Path,
-  private val forcedUpload: Boolean,
+internal suspend fun uploadJpsCache(
+  forcedUpload: Boolean,
+  commitHash: String,
+  authHeader: CharSequence,
+  s3Dir: Path?,
+  uploadUrl: URI,
+  context: CompilationContext,
 ) {
-  init {
-    s3Folder.deleteRecursively()
-    Files.createDirectories(s3Folder)
+  if (s3Dir != null) {
+    s3Dir.deleteRecursively()
+    Files.createDirectories(s3Dir)
  }

-  suspend fun upload(messages: BuildMessages) {
-    val sourceStateFile = context.compilationData.dataStorageRoot.resolve(SOURCES_STATE_FILE_NAME)
-
-    val sourceState = try {
-      Files.newBufferedReader(sourceStateFile).use {
-        BuildTargetSourcesState.readJson(JsonReader(it))
-      }
-    }
-    catch (e: NoSuchFileException) {
-      throw IllegalStateException(
-        "Compilation outputs doesn't contain source state file, " +
-        "please enable '${ProjectStamps.PORTABLE_CACHES_PROPERTY}' flag", e
-      )
+  val sourceStateFile = context.compilationData.dataStorageRoot.resolve(SOURCES_STATE_FILE_NAME)
+  val sourceState = try {
+    Files.newBufferedReader(sourceStateFile).use {
+      BuildTargetSourcesState.readJson(JsonReader(it))
    }
+  }
+  catch (e: NoSuchFileException) {
+    throw IllegalStateException("Compilation outputs doesn't contain source state file, please enable '${ProjectStamps.PORTABLE_CACHES_PROPERTY}' flag", e)
+  }

-    val start = System.nanoTime()
-    val totalUploadedBytes = AtomicLong()
-    val uploadedOutputCount = AtomicInteger()
-    val uploader = Uploader(remoteCache.uploadUrl, remoteCache.authHeader)
-    withContext(Dispatchers.IO) {
-      // Jps Caches upload is started first because of significant size
-      launch {
-        spanBuilder("upload jps cache").use {
-          uploadJpsCaches(uploader)
+  val start = System.nanoTime()
+  val totalUploadedBytes = LongAdder()
+  val uploadedOutputCount = LongAdder()
+  val messages = context.messages
+  withHttp2ClientConnectionFactory(trustAll = uploadUrl.host == "127.0.0.1") { client ->
+    client.connect(uploadUrl.host, uploadUrl.port).withAuth(authHeader).use { connection ->
+      val urlPathPrefix = uploadUrl.path
+      withContext(Dispatchers.IO) {
+        launch {
+          spanBuilder("upload jps cache").use {
+            uploadJpsCaches(urlPathPrefix = urlPathPrefix, connection = connection, s3Dir = s3Dir, forcedUpload = forcedUpload, commitHash = commitHash, context = context)
+          }
+        }
+
+        spanBuilder("upload compilation outputs").use {
+          val allCompilationOutputs = getAllCompilationOutputs(sourceState = sourceState, classOutDir = context.classesOutputDirectory)
+          val tempDir = Files.createTempDirectory(context.paths.tempDir, "jps-cache-bytecode-")
+          try {
+            uploadCompilationOutputs(
+              uploadedOutputCount = uploadedOutputCount,
+              totalUploadedBytes = totalUploadedBytes,
+              allCompilationOutputs = allCompilationOutputs,
+              urlPathPrefix = urlPathPrefix,
+              connection = connection,
+              s3Dir = s3Dir,
+              forcedUpload = forcedUpload,
+              tempDir = tempDir,
+            )
+          }
+          finally {
+            tempDir.deleteRecursively()
+          }
+          messages.reportStatisticValue("Total outputs", allCompilationOutputs.size.toString())
        }
      }

-      spanBuilder("upload compilation outputs").use {
-        val allCompilationOutputs = getAllCompilationOutputs(sourceState = sourceState, classOutDir = context.classesOutputDirectory)
-        uploadCompilationOutputs(
-          uploader = uploader,
-          uploadedOutputCount = uploadedOutputCount,
-          allCompilationOutputs = allCompilationOutputs,
-        )
-        messages.reportStatisticValue("Total outputs", allCompilationOutputs.size.toString())
-      }
-    }
-
-    messages.reportStatisticValue("jps-cache:upload:time", TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start).toString())
-    messages.reportStatisticValue("jps-cache:uploaded:count", (uploadedOutputCount.get() + 1).toString())
-    messages.reportStatisticValue("jps-cache:uploaded:bytes", "$totalUploadedBytes")
-
-    messages.reportStatisticValue("Uploaded outputs", uploadedOutputCount.get().toString())
-
-    uploadMetadata(sourceStateFile = sourceStateFile, uploader = uploader)
-    uploadToS3()
-  }
-
-  private suspend fun uploadToS3() {
-    spanBuilder("aws s3 sync").use {
-      awsS3Cli("cp", "--no-progress", "--include", "*", "--recursive", s3Folder.toString(), "s3://intellij-jps-cache", returnStdOut = false)
-    }
-  }
-
-  private suspend fun uploadJpsCaches(uploader: Uploader) {
-    val dataStorageRoot = context.compilationData.dataStorageRoot
-    val zipFile = dataStorageRoot.parent.resolve(commitHash)
-    Compressor.Zip(zipFile).use { zip ->
-      zip.addDirectory(dataStorageRoot)
-    }
-    val cachePath = "caches/$commitHash"
-    if (forcedUpload || !uploader.isExist(cachePath, true)) {
-      uploader.upload(cachePath, zipFile)
-    }
-    moveFile(zipFile, s3Folder.resolve(cachePath))
-  }
-
-  private suspend fun uploadMetadata(sourceStateFile: Path, uploader: Uploader) {
-    val metadataPath = "metadata/$commitHash"
-    spanBuilder("upload metadata").setAttribute("path", metadataPath).use {
-      uploader.upload(metadataPath, sourceStateFile)
-      copyFile(sourceStateFile, s3Folder.resolve(metadataPath))
-    }
-  }
-
-  private suspend fun uploadCompilationOutputs(
-    uploader: Uploader,
-    uploadedOutputCount: AtomicInteger,
-    allCompilationOutputs: List<CompilationOutput>,
-  ) {
-    allCompilationOutputs.forEachConcurrent(uploadParallelism) { compilationOutput ->
-      val outDir = compilationOutput.path
-      spanBuilder("upload output part")
-        .setAttribute("part", compilationOutput.remotePath)
-        .setAttribute("path", outDir.toString())
-        .use { span ->
-          val sourcePath = compilationOutput.remotePath
-          if (Files.notExists(outDir)) {
-            span.addEvent("doesn't exist, was a respective module removed?")
-            return@use
-          }
-
-          val zipFile = context.paths.tempDir.resolve("compilation-output-zips").resolve(sourcePath)
-          zipWithCompression(zipFile, mapOf(outDir to ""))
-          if (forcedUpload || !uploader.isExist(sourcePath)) {
-            uploader.upload(sourcePath, zipFile)
-            uploadedOutputCount.incrementAndGet()
-          }
-          moveFile(zipFile, s3Folder.resolve(sourcePath))
+      val metadataPath = "metadata/$commitHash"
+      val metadataUrl = "$urlPathPrefix/$metadataPath"
+      spanBuilder("upload metadata").setAttribute("path", metadataUrl).use {
+        connection.upload(metadataUrl, sourceStateFile)
+        if (s3Dir != null) {
+          copyFile(file = sourceStateFile, target = s3Dir.resolve(metadataPath))
        }
-    }
-  }
-
-  /**
-   * Upload and publish a file with commits history
-   */
-  internal suspend fun updateCommitHistory(overrideCommits: Set<String>? = null) {
-    val overrideRemoteHistory = overrideCommits != null
-    val commitHistory = CommitsHistory(mapOf(remoteGitUrl to (overrideCommits ?: setOf(commitHash))))
-    val uploader = Uploader(serverUrl = remoteCache.uploadUrl, authHeader = remoteCache.authHeader)
-    for (commitHash in commitHistory.commitsForRemote(remoteGitUrl)) {
-      val cacheUploaded = uploader.isExist("caches/$commitHash")
-      val metadataUploaded = uploader.isExist("metadata/$commitHash")
-      if (!cacheUploaded && !metadataUploaded) {
-        val msg = "Unable to publish $commitHash due to missing caches/$commitHash and metadata/$commitHash. " +
-                  "Probably caused by previous cleanup build."
-        if (overrideRemoteHistory) context.messages.error(msg) else context.messages.warning(msg)
-        return
-      }
-      check(cacheUploaded == metadataUploaded) {
-        "JPS Caches are uploaded: $cacheUploaded, metadata is uploaded: $metadataUploaded"
      }
    }
-    val newHistory = if (overrideRemoteHistory) commitHistory else commitHistory + remoteCommitHistory(uploader)
-    uploader.upload(path = CommitsHistory.JSON_FILE, file = writeCommitHistory(newHistory))
-    val expected = newHistory.commitsForRemote(remoteGitUrl).toSet()
-    val actual = remoteCommitHistory(uploader).commitsForRemote(remoteGitUrl).toSet()
-    val missing = expected - actual
-    val unexpected = actual - expected
-    check(missing.none() && unexpected.none()) {
-      """
-        Missing: $missing
-        Unexpected: $unexpected
-      """.trimIndent()
-    }
  }

-  private suspend fun remoteCommitHistory(uploader: Uploader): CommitsHistory {
-    return if (uploader.isExist(CommitsHistory.JSON_FILE)) {
-      CommitsHistory(uploader.getAsString(CommitsHistory.JSON_FILE, remoteCache.authHeader))
-    }
-    else {
-      CommitsHistory(emptyMap())
-    }
-  }
+  messages.reportStatisticValue("jps-cache:upload:time", TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start).toString())
+  messages.reportStatisticValue("jps-cache:uploaded:count", (uploadedOutputCount.sum() + 1).toString())
+  messages.reportStatisticValue("jps-cache:uploaded:bytes", totalUploadedBytes.sum().toString())

-  private fun writeCommitHistory(commitHistory: CommitsHistory): Path {
-    val commitHistoryFile = s3Folder.resolve(CommitsHistory.JSON_FILE)
-    Files.createDirectories(commitHistoryFile.parent)
-    val json = commitHistory.toJson()
-    Files.writeString(commitHistoryFile, json)
-    Span.current().addEvent("write commit history", Attributes.of(AttributeKey.stringKey("data"), json))
-    return commitHistoryFile
+  messages.reportStatisticValue("Uploaded outputs", uploadedOutputCount.sum().toString())
+
+  if (s3Dir != null) {
+    uploadToS3(s3Dir)
  }
 }

-private class Uploader(serverUrl: String, @JvmField val authHeader: String) {
-  private val serverUrl = serverUrl.trimEnd('/')
+private suspend fun uploadToS3(s3Dir: Path) {
+  spanBuilder("aws s3 sync").use {
+    awsS3Cli("cp", "--no-progress", "--include", "*", "--recursive", s3Dir.toString(), "s3://intellij-jps-cache", returnStdOut = false)
+  }
+}

-  suspend fun upload(path: String, file: Path) {
-    val url = pathToUrl(path)
-    spanBuilder("upload").setAttribute("url", url).setAttribute("path", path).use {
-      val fileSize = Files.size(file)
-      retryWithExponentialBackOff {
-        httpClient.newCall(Request.Builder().url(url)
-          .header("Authorization", authHeader)
-          .put(object : RequestBody() {
-            override fun contentType() = MEDIA_TYPE_BINARY
+private suspend fun uploadJpsCaches(
+  urlPathPrefix: String,
+  connection: Http2ClientConnection,
+  commitHash: String,
+  forcedUpload: Boolean,
+  s3Dir: Path?,
+  context: CompilationContext,
+) {
+  val dataStorageRoot = context.compilationData.dataStorageRoot
+  val zipFile = context.paths.tempDir.resolve("$commitHash-${java.lang.Long.toUnsignedString(Random.nextLong(), Character.MAX_RADIX)}.zip")
+  try {
+    val compressed by lazy {
+      zipWithCompression(zipFile, mapOf(dataStorageRoot to ""))
+      zipFile
+    }

-            override fun contentLength(): Long = fileSize
+    val cachePath = "caches/$commitHash"
+    val urlPath = "$urlPathPrefix/$cachePath"
+    if (forcedUpload || !checkExists(connection, urlPath, logIfExists = true)) {
+      connection.upload(path = urlPath, file = compressed)
+    }

-            override fun writeTo(sink: BufferedSink) {
-              file.source().use {
-                sink.writeAll(it)
-              }
-            }
-          }).build()).executeAsync().useSuccessful {}
-      }
+    if (s3Dir != null) {
+      moveFile(compressed, s3Dir.resolve(cachePath))
    }
  }
+  finally {
+    Files.deleteIfExists(zipFile)
+  }
+}

-  suspend fun isExist(path: String, logIfExists: Boolean = false): Boolean {
-    val url = pathToUrl(path)
-    return spanBuilder("head").setAttribute("url", url).use { span ->
-      val code = retryWithExponentialBackOff {
-        httpClient.head(url, authHeader)
-      }
+private suspend fun uploadCompilationOutputs(
+  uploadedOutputCount: LongAdder,
+  allCompilationOutputs: List<CompilationOutput>,
+  urlPathPrefix: String,
+  connection: Http2ClientConnection,
+  totalUploadedBytes: LongAdder,
+  forcedUpload: Boolean,
+  tempDir: Path,
+  s3Dir: Path?,
+) {
+  allCompilationOutputs.forEachConcurrent(uploadParallelism) { compilationOutput ->
+    val outDir = compilationOutput.path
+    spanBuilder("upload output part")
+      .setAttribute("part", compilationOutput.remotePath)
+      .setAttribute("path", outDir.toString())
+      .use { span ->
+        if (Files.notExists(outDir)) {
+          span.addEvent("doesn't exist, was a respective module removed?")
+          return@use
+        }

-      if (code != 200) {
-        return@use false
-      }
+        val sourcePath = compilationOutput.remotePath
+        val zipFile = tempDir.resolve("${sourcePath.replace('/', '_')}.zip")
+        zipWithCompression(targetFile = zipFile, dirs = mapOf(outDir to ""), createFileParentDirs = false)

-      try {
-        /**
-         * FIXME dirty workaround for unreliable [serverUrl]
-         */
-        httpClient.get(url, authHeader) {
-          it.peekBody(byteCount = 1)
+        val urlPath = "$urlPathPrefix/$sourcePath"
+        if (forcedUpload || !checkExists(connection, urlPath)) {
+          spanBuilder("upload").setAttribute("urlPath", urlPath).setAttribute("path", sourcePath).use {
+            val result = connection.upload(path = urlPath, file = zipFile)
+            totalUploadedBytes.add(result.uploadedSize)
+          }
+          uploadedOutputCount.increment()
+        }
+
+        if (s3Dir != null) {
+          moveFile(zipFile, s3Dir.resolve(sourcePath))
        }
      }
-      catch (_: Exception) {
-        return@use false
+  }
+}
+
+/**
+ * Upload and publish a file with commits history
+ */
+internal suspend fun updateJpsCacheCommitHistory(
+  overrideCommits: Set<String>?,
+  remoteGitUrl: String,
+  commitHash: String,
+  uploadUrl: URI,
+  authHeader: CharSequence,
+  s3Dir: Path?,
+  context: CompilationContext,
+) {
+  val overrideRemoteHistory = overrideCommits != null
+  val commitHistory = CommitsHistory(mapOf(remoteGitUrl to (overrideCommits ?: setOf(commitHash))))
+  withHttp2ClientConnectionFactory(trustAll = uploadUrl.host == "127.0.0.1") { client ->
+    val urlPathPrefix = uploadUrl.path
+    client.connect(uploadUrl.host, uploadUrl.port).withAuth(authHeader).use { connection ->
+      for (commitHashForRemote in commitHistory.commitsForRemote(remoteGitUrl)) {
+        val cacheUploaded = checkExists(connection, "$urlPathPrefix/caches/$commitHashForRemote")
+        val metadataUploaded = checkExists(connection, "$urlPathPrefix/metadata/$commitHashForRemote")
+        if (!cacheUploaded && !metadataUploaded) {
+          val msg = "Unable to publish $commitHashForRemote due to missing caches/$commitHashForRemote and metadata/$commitHashForRemote." +
+                    " Probably caused by previous cleanup build."
+          if (overrideRemoteHistory) {
+            context.messages.error(msg)
+          }
+          else {
+            context.messages.warning(msg)
+          }
+          return@use false
+        }
+        check(cacheUploaded == metadataUploaded) {
+          "JPS Caches are uploaded: $cacheUploaded, metadata is uploaded: $metadataUploaded"
+        }
      }

-      if (logIfExists) {
-        span.addEvent("File '$path' already exists on server, nothing to upload")
+      val newHistory = if (overrideRemoteHistory) commitHistory else commitHistory + remoteCommitHistory(connection, urlPathPrefix)
+      connection.upload(path = "$urlPathPrefix/${CommitsHistory.JSON_FILE}", file = writeCommitHistory(commitHistory = newHistory, context = context, s3Dir = s3Dir))
+      val expected = newHistory.commitsForRemote(remoteGitUrl).toSet()
+      val actual = remoteCommitHistory(connection, urlPathPrefix).commitsForRemote(remoteGitUrl).toSet()
+      val missing = expected - actual
+      val unexpected = actual - expected
+      check(missing.none() && unexpected.none()) {
+        """
+          Missing: $missing
+          Unexpected: $unexpected
+        """.trimIndent()
      }
-      true
    }
  }
+}

-  suspend fun getAsString(path: String, authHeader: String) = retryWithExponentialBackOff {
-    httpClient.get(pathToUrl(path), authHeader) { it.body.string() }
+private suspend fun remoteCommitHistory(connection: Http2ClientConnection, urlPathPrefix: String): CommitsHistory {
+  return CommitsHistory(connection.getJsonOrDefaultIfNotFound(path = "$urlPathPrefix/${CommitsHistory.JSON_FILE}", defaultIfNotFound = emptyMap()))
+}
+
+private fun writeCommitHistory(commitHistory: CommitsHistory, context: CompilationContext, s3Dir: Path?): Path {
+  val commitHistoryFile = (s3Dir ?: context.paths.tempDir).resolve(CommitsHistory.JSON_FILE)
+  Files.createDirectories(commitHistoryFile.parent)
+  val json = commitHistory.toJson()
+  Files.writeString(commitHistoryFile, json)
+  Span.current().addEvent("write commit history", Attributes.of(AttributeKey.stringKey("data"), json))
+  return commitHistoryFile
+}
+
+private suspend fun checkExists(
+  connection: Http2ClientConnection,
+  urlPath: String,
+  logIfExists: Boolean = false,
+): Boolean {
+  val status = connection.head(urlPath)
+  if (status == HttpResponseStatus.OK) {
+    // already exists
+    if (logIfExists) {
+      Span.current().addEvent("File $urlPath already exists on server, nothing to upload")
+    }
+    return true
  }
-
-  private fun pathToUrl(path: String) = "$serverUrl/${path.trimStart('/')}"
+  else if (status != HttpResponseStatus.NOT_FOUND) {
+    Span.current().addEvent("unexpected response status for HEAD request", Attributes.of(AttributeKey.stringKey("status"), status.toString()))
+  }
+  return false
 }
@@ -1,19 +1,18 @@
-// Copyright 2000-2022 JetBrains s.r.o. and contributors. Use of this source code is governed by the Apache 2.0 license.
+// Copyright 2000-2024 JetBrains s.r.o. and contributors. Use of this source code is governed by the Apache 2.0 license.
 package org.jetbrains.intellij.build.impl.compilation.cache

-import com.google.gson.Gson
-import com.google.gson.reflect.TypeToken
+import kotlinx.serialization.encodeToString
+import kotlinx.serialization.json.Json

 class CommitsHistory(private val commitsPerRemote: Map<String, Set<String>>) {
  companion object {
-    const val JSON_FILE = "commit_history.json"
-    private val JSON_TYPE = object : TypeToken<Map<String, Set<String>>>() {}.type
+    internal const val JSON_FILE = "commit_history.json"
  }

-  constructor(json: String) : this(Gson().fromJson(json, JSON_TYPE) as Map<String, Set<String>>)
+  constructor(json: String) : this(Json.decodeFromString<Map<String, Set<String>>>(json))

  fun toJson(): String {
-    return Gson().toJson(commitsPerRemote)
+    return Json.encodeToString(commitsPerRemote)
  }

  fun commitsForRemote(remote: String): Collection<String> {
@@ -1,12 +1,10 @@
 // Copyright 2000-2024 JetBrains s.r.o. and contributors. Use of this source code is governed by the Apache 2.0 license.
 package org.jetbrains.intellij.build.impl.compilation.cache

-import com.google.gson.stream.JsonReader
 import org.jetbrains.intellij.build.impl.compilation.CompilationOutput
 import org.jetbrains.jps.builders.java.JavaModuleBuildTargetType
 import org.jetbrains.jps.builders.java.ResourcesTargetType
 import org.jetbrains.jps.cache.model.BuildTargetState
-import org.jetbrains.jps.incremental.storage.BuildTargetSourcesState
 import java.nio.file.Path

 private const val IDENTIFIER = "\$BUILD_DIR\$"
@@ -23,10 +21,6 @@ internal fun getAllCompilationOutputs(sourceState: Map<String, Map<String, Build
  return getProductionCompilationOutputs(sourceState, classOutDir) + getTestsCompilationOutputs(sourceState, classOutDir)
 }

-internal fun parseSourcesStateFile(json: String): Map<String, Map<String, BuildTargetState>> {
-  return BuildTargetSourcesState.readJson(JsonReader(json.reader()))
-}
-
 private fun getProductionCompilationOutputs(currentSourcesState: Map<String, Map<String, BuildTargetState>>, classOutDir: Path): List<CompilationOutput> {
  return getCompilationOutputs(
    // production modules with both classes and resources
@@ -0,0 +1,37 @@
+// Copyright 2000-2024 JetBrains s.r.o. and contributors. Use of this source code is governed by the Apache 2.0 license.
+package org.jetbrains.intellij.build.impl.compilation
+
+import org.jetbrains.intellij.build.http2Client.READ_OPERATION
+import java.math.BigInteger
+import java.nio.ByteBuffer
+import java.nio.channels.FileChannel
+import java.nio.file.Path
+import java.security.MessageDigest
+
+private val sharedDigest = MessageDigest.getInstance("SHA-256", java.security.Security.getProvider("SUN"))
+internal fun sha256() = sharedDigest.clone() as MessageDigest
+
+internal fun computeHash(file: Path): String {
+  val messageDigest = sha256()
+  FileChannel.open(file, READ_OPERATION).use { channel ->
+    val fileSize = channel.size()
+    // java message digest doesn't support native buffer (copies to a heap byte array in any case)
+    val bufferSize = 256 * 1024
+    val buffer = ByteBuffer.allocate(bufferSize)
+    var offset = 0L
+    var readBytes: Int
+    while (offset < fileSize) {
+      buffer.clear()
+      readBytes = channel.read(buffer, offset)
+      if (readBytes <= 0) {
+        break
+      }
+
+      messageDigest.update(buffer.array(), 0, readBytes)
+      offset += readBytes
+    }
+  }
+  return digestToString(messageDigest)
+}
+
+internal fun digestToString(digest: MessageDigest): String = BigInteger(1, digest.digest()).toString(36) + "z"
@@ -1,7 +1,6 @@
 // Copyright 2000-2024 JetBrains s.r.o. and contributors. Use of this source code is governed by the Apache 2.0 license.
 package org.jetbrains.intellij.build.impl.compilation

-import com.intellij.util.lang.HashMapZipFile
 import io.opentelemetry.api.common.AttributeKey
 import io.opentelemetry.api.common.Attributes
 import kotlinx.coroutines.isActive
@@ -11,21 +10,14 @@ import org.jetbrains.intellij.build.http2Client.Http2ClientConnection
 import org.jetbrains.intellij.build.http2Client.Http2ClientConnectionFactory
 import org.jetbrains.intellij.build.http2Client.ZstdDecompressContextPool
 import org.jetbrains.intellij.build.http2Client.download
-import org.jetbrains.intellij.build.io.INDEX_FILENAME
 import org.jetbrains.intellij.build.telemetry.TraceManager.spanBuilder
 import org.jetbrains.intellij.build.telemetry.use
 import java.net.URI
-import java.nio.channels.FileChannel
 import java.nio.file.Files
-import java.nio.file.Path
-import java.nio.file.StandardOpenOption
-import java.util.*
 import java.util.concurrent.CancellationException
 import java.util.concurrent.atomic.AtomicLong
 import kotlin.io.path.name

-private val OVERWRITE_OPERATION = EnumSet.of(StandardOpenOption.WRITE, StandardOpenOption.CREATE, StandardOpenOption.TRUNCATE_EXISTING)
-
 internal suspend fun downloadCompilationCache(
  serverUrl: String,
  client: Http2ClientConnectionFactory,
@@ -35,38 +27,10 @@ internal suspend fun downloadCompilationCache(
  skipUnpack: Boolean,
  saveHash: Boolean,
 ) {
-  var urlPathWithPrefix = "/$prefix/"
-  // first let's check for initial redirect (mirror selection)
-  val initialServerUri = URI(serverUrl)
-  var effectiveServerUri = initialServerUri
-  var connection: Http2ClientConnection? = client.connect(effectiveServerUri.host, effectiveServerUri.port)
-  try {
-    spanBuilder("mirror selection").use { span ->
-      val newLocation = connection!!.getRedirectLocation(urlPathWithPrefix)
-      if (newLocation == null) {
-        span.addEvent("origin server will be used", Attributes.of(AttributeKey.stringKey("url"), urlPathWithPrefix))
-      }
-      else {
-        effectiveServerUri = URI(newLocation.toString())
-        urlPathWithPrefix = effectiveServerUri.path
-        span.addEvent("redirected to mirror", Attributes.of(AttributeKey.stringKey("url"), urlPathWithPrefix))
-      }
-    }
-  }
-  finally {
-    if (initialServerUri != effectiveServerUri) {
-      connection?.close()
-      connection = null
-    }
-  }
-
-  if (connection == null) {
-    connection = client.connect(effectiveServerUri.host, effectiveServerUri.port)
-  }
-  try {
+  checkMirrorAndConnect(serverUrl = serverUrl, client = client, prefix = prefix) { connection, urlPathPrefix ->
    val zstdDecompressContextPool = ZstdDecompressContextPool()
    toDownload.forEachConcurrent(downloadParallelism) { item ->
-      val urlPath = "$urlPathWithPrefix${item.name}/${item.file.fileName}"
+      val urlPath = "$urlPathPrefix/${item.name}/${item.file.fileName}"
      spanBuilder("download").setAttribute("name", item.name).setAttribute("urlPath", urlPath).use { span ->
        try {
          downloadedBytes.getAndAdd(
@@ -93,8 +57,46 @@ internal suspend fun downloadCompilationCache(
      }
    }
  }
+}
+
+internal suspend fun checkMirrorAndConnect(
+  serverUrl: String,
+  client: Http2ClientConnectionFactory,
+  prefix: String,
+  block: suspend (connection: Http2ClientConnection, urlPathPrefix: String) -> Unit,
+) {
+  var urlPathWithSlash = "/$prefix/"
+  // first let's check for initial redirect (mirror selection)
+  val initialServerUri = URI(serverUrl)
+  var effectiveServerUri = initialServerUri
+  var connection: Http2ClientConnection? = client.connect(effectiveServerUri.host, effectiveServerUri.port)
+  var connectionToClose = connection
+  try {
+    spanBuilder("mirror selection").use { span ->
+      val newLocation = connection!!.getRedirectLocation(urlPathWithSlash)
+      if (newLocation == null) {
+        span.addEvent("origin server will be used", Attributes.of(AttributeKey.stringKey("url"), urlPathWithSlash))
+        connectionToClose = null
+      }
+      else {
+        effectiveServerUri = URI(newLocation.toString())
+        urlPathWithSlash = effectiveServerUri.path
+        span.addEvent("redirected to mirror", Attributes.of(AttributeKey.stringKey("url"), urlPathWithSlash))
+        connection = null
+      }
+    }
+  }
  finally {
-    connection.close()
+    connectionToClose?.close()
+    connectionToClose = null
+  }
+
+  val effectiveConnection = connection ?: client.connect(effectiveServerUri.host, effectiveServerUri.port)
+  try {
+    block(effectiveConnection, urlPathWithSlash.trimEnd('/'))
+  }
+  finally {
+    effectiveConnection.close()
  }
 }

@@ -123,7 +125,7 @@ private suspend fun download(

  if (!skipUnpack) {
    spanBuilder("unpack").setAttribute("name", item.name).use {
-      unpackArchive(item, saveHash)
+      unpackCompilationPartArchive(item, saveHash)
    }
  }
  return downloaded
@@ -135,29 +137,8 @@ internal class CompilePartDownloadFailedError(@JvmField val item: FetchAndUnpack

 internal class HashMismatchException(message: String) : IOException(message)

-internal fun unpackArchive(item: FetchAndUnpackItem, saveHash: Boolean) {
-  HashMapZipFile.load(item.file).use { zipFile ->
-    val root = item.output
-    Files.createDirectories(root)
-    val createdDirs = HashSet<Path>()
-    createdDirs.add(root)
-    for (entry in zipFile.entries) {
-      if (entry.isDirectory || entry.name == INDEX_FILENAME) {
-        continue
-      }
-
-      val file = root.resolve(entry.name)
-      val parent = file.parent
-      if (createdDirs.add(parent)) {
-        Files.createDirectories(parent)
-      }
-
-      FileChannel.open(file, OVERWRITE_OPERATION).use { channel ->
-        channel.write(entry.getByteBuffer(zipFile), 0)
-      }
-    }
-  }
-
+internal fun unpackCompilationPartArchive(item: FetchAndUnpackItem, saveHash: Boolean) {
+  unpackArchiveUsingNettyByteBufferPool(archiveFile = item.file, outDir = item.output, isCompressed = false)
  if (saveHash) {
    // save actual hash
    Files.writeString(item.output.resolve(".hash"), item.hash)
@@ -5,34 +5,10 @@ import kotlinx.coroutines.ExperimentalCoroutinesApi
 import kotlinx.coroutines.suspendCancellableCoroutine
 import okhttp3.*
 import okhttp3.internal.closeQuietly
-import org.jetbrains.intellij.build.NoMoreRetriesException
 import java.io.IOException
 import java.util.concurrent.TimeUnit
 import kotlin.coroutines.resumeWithException

-internal suspend fun OkHttpClient.head(url: String, authHeader: String): Int {
-  return newCall(Request.Builder().url(url).head().header("Authorization", authHeader).build()).executeAsync().use { response ->
-    if (response.code != 200 && response.code != 404) {
-      throw IOException("Unexpected code $response")
-    }
-    response.code
-  }
-}
-
-internal suspend fun <T> OkHttpClient.get(url: String, authHeader: String, task: (Response) -> T): T {
-  return newCall(Request.Builder().url(url).header("Authorization", authHeader).build()).executeAsync().useSuccessful(task)
-}
-
-internal inline fun <T> Response.useSuccessful(task: (Response) -> T): T {
-  return use { response ->
-    when {
-      response.isSuccessful -> task(response)
-      response.code == 404 -> throw NoMoreRetriesException("Unexpected code $response")
-      else -> throw IOException("Unexpected code $response")
-    }
-  }
-}
-
 internal val httpClient: OkHttpClient by lazy {
  val timeout = 1L
  val unit = TimeUnit.MINUTES
@@ -79,7 +55,7 @@ internal val httpClient: OkHttpClient by lazy {
    .build()
 }

-@ExperimentalCoroutinesApi // resume with a resource cleanup.
+@ExperimentalCoroutinesApi
 internal suspend fun Call.executeAsync(): Response {
  return suspendCancellableCoroutine { continuation ->
    continuation.invokeOnCancellation {
@@ -13,6 +13,7 @@ import kotlinx.serialization.Serializable
 import org.jetbrains.intellij.build.forEachConcurrent
 import org.jetbrains.intellij.build.http2Client.Http2ClientConnection
 import org.jetbrains.intellij.build.http2Client.ZstdCompressContextPool
+import org.jetbrains.intellij.build.http2Client.post
 import org.jetbrains.intellij.build.http2Client.upload
 import org.jetbrains.intellij.build.telemetry.TraceManager.spanBuilder
 import org.jetbrains.intellij.build.telemetry.use
@@ -0,0 +1,57 @@
+// Copyright 2000-2024 JetBrains s.r.o. and contributors. Use of this source code is governed by the Apache 2.0 license.
+package org.jetbrains.intellij.build.impl.compilation
+
+import com.intellij.util.lang.HashMapZipFile
+import io.netty.buffer.ByteBuf
+import io.netty.buffer.PooledByteBufAllocator
+import org.jetbrains.intellij.build.io.INDEX_FILENAME
+import java.nio.channels.FileChannel
+import java.nio.file.Files
+import java.nio.file.Path
+import java.nio.file.StandardOpenOption
+import java.util.*
+import java.util.function.IntFunction
+
+private val OVERWRITE_OPERATION = EnumSet.of(StandardOpenOption.WRITE, StandardOpenOption.CREATE, StandardOpenOption.TRUNCATE_EXISTING)
+
+internal fun unpackArchiveUsingNettyByteBufferPool(archiveFile: Path, outDir: Path, isCompressed: Boolean) {
+  Files.createDirectories(outDir)
+  HashMapZipFile.load(archiveFile).use { zipFile ->
+    val createdDirs = HashSet<Path>()
+    createdDirs.add(outDir)
+    for (entry in zipFile.entries) {
+      if (entry.isDirectory || entry.name == INDEX_FILENAME) {
+        continue
+      }
+
+      val file = outDir.resolve(entry.name)
+      val parent = file.parent
+      if (createdDirs.add(parent)) {
+        Files.createDirectories(parent)
+      }
+
+      FileChannel.open(file, OVERWRITE_OPERATION).use { channel ->
+        if (isCompressed) {
+          // we use netty buffer as a part of compilation cache/parts tasks - reuse it
+          var nettyBuffer: ByteBuf? = null
+          try {
+            val byteBuffer = entry.getByteBuffer(
+              zipFile,
+              IntFunction { size ->
+                val buffer = PooledByteBufAllocator.DEFAULT.directBuffer(size)
+                nettyBuffer = buffer
+                buffer.nioBuffer(0, size)
+              })
+            channel.write(byteBuffer, 0)
+          }
+          finally {
+            nettyBuffer?.release()
+          }
+        }
+        else {
+          channel.write(entry.getByteBuffer(zipFile, null))
+        }
+      }
+    }
+  }
+}
@@ -1,11 +1,10 @@
-// Copyright 2000-2023 JetBrains s.r.o. and contributors. Use of this source code is governed by the Apache 2.0 license.
+// Copyright 2000-2024 JetBrains s.r.o. and contributors. Use of this source code is governed by the Apache 2.0 license.
@file:Suppress("ReplacePutWithAssignment", "ReplaceGetOrSet", "HardCodedStringLiteral")

 package org.jetbrains.intellij.build.impl.logging.jps

 import com.intellij.openapi.diagnostic.DefaultLogger
 import com.intellij.openapi.diagnostic.Logger
-import com.intellij.util.containers.MultiMap
 import com.jetbrains.plugin.structure.base.utils.createParentDirs
 import io.opentelemetry.api.common.AttributeKey
 import io.opentelemetry.api.common.Attributes
@@ -23,11 +22,12 @@ import java.nio.file.Files
 import java.util.*
 import java.util.concurrent.CancellationException
 import java.util.concurrent.ConcurrentHashMap
+import java.util.concurrent.CopyOnWriteArrayList
 import java.util.concurrent.TimeUnit
 import java.util.function.BiConsumer

@Internal
-suspend fun withJpsLogging(context: CompilationContext, span: Span, action: suspend (JpsMessageHandler) -> Unit) {
+internal suspend fun withJpsLogging(context: CompilationContext, span: Span, action: suspend (JpsMessageHandler) -> Unit) {
  val messageHandler = JpsMessageHandler(context, span)
  JpsLoggerFactory.messageHandler = messageHandler
  if (context.options.compilationLogEnabled) {
@@ -69,7 +69,9 @@ suspend fun withJpsLogging(context: CompilationContext, span: Span, action: susp
@VisibleForTesting
 class JpsLoggerFactory : Logger.Factory {
  companion object {
-    var messageHandler: JpsMessageHandler? = null
+    @JvmField
+    internal var messageHandler: JpsMessageHandler? = null
+    @JvmField
    var fileLoggerFactory: JpsFileLoggerFactory? = null
  }

@@ -78,12 +80,12 @@ class JpsLoggerFactory : Logger.Factory {
  }
 }

-@Internal
-class JpsMessageHandler(private val context: CompilationContext, private val span: Span) : MessageHandler {
-  val errorMessagesByCompiler = MultiMap.createConcurrent<String, String>()
+internal class JpsMessageHandler(private val context: CompilationContext, private val span: Span) : MessageHandler {
+  @JvmField
+  val errorMessagesByCompiler = ConcurrentHashMap<String, MutableList<String>>()
  private val compilationStartTimeForTarget = ConcurrentHashMap<String, Long>()
  private val compilationFinishTimeForTarget = ConcurrentHashMap<String, Long>()
-  private var progress = (-1.0).toFloat()
+  private var progress = -1.0f

  override fun processMessage(message: BuildMessage): Unit = TeamCityBuildMessageLogger.withFlow(span) {
    val text = message.messageText
@@ -97,8 +99,10 @@ class JpsMessageHandler(private val context: CompilationContext, private val spa
          messageText = buildString {
            if (sourcePath != null) {
              append(sourcePath)
-              if (message.line != -1L) append(":" + message.line)
-              appendLine(":")
+              if (message.line != -1L) {
+                append(':').append(message.line)
+              }
+              appendLine(':')
            }
            append(text)
            val moduleNames = message.moduleNames
@@ -111,7 +115,7 @@ class JpsMessageHandler(private val context: CompilationContext, private val spa
          compilerName = ""
          messageText = text
        }
-        errorMessagesByCompiler.putValue(compilerName, messageText)
+        errorMessagesByCompiler.computeIfAbsent(compilerName) { CopyOnWriteArrayList() }.add(messageText)
      }
      BuildMessage.Kind.WARNING -> context.messages.warning(text)
      BuildMessage.Kind.INFO, BuildMessage.Kind.JPS_INFO -> if (message is BuilderStatisticsMessage) {
@@ -1,4 +1,4 @@
-// Copyright 2000-2023 JetBrains s.r.o. and contributors. Use of this source code is governed by the Apache 2.0 license.
+// Copyright 2000-2024 JetBrains s.r.o. and contributors. Use of this source code is governed by the Apache 2.0 license.
 package com.intellij.util.lang;

 import org.jetbrains.annotations.ApiStatus;
@@ -96,7 +96,7 @@ public final class HashMapZipFile implements ZipFile {
  @Override
  public ByteBuffer getByteBuffer(@NotNull String path) throws IOException {
    ImmutableZipEntry entry = getRawEntry(path.charAt(0) == '/' ? path.substring(1) : path);
-    return entry == null ? null : entry.getByteBuffer(this);
+    return entry == null ? null : entry.getByteBuffer(this, null);
  }

  /**
@@ -122,7 +122,7 @@ public final class HashMapZipFile implements ZipFile {

      @Override
      public @NotNull ByteBuffer getByteBuffer() throws IOException {
-        return entry.getByteBuffer(HashMapZipFile.this);
+        return entry.getByteBuffer(HashMapZipFile.this, null);
      }

      @Override
@@ -1,13 +1,15 @@
-// Copyright 2000-2021 JetBrains s.r.o. and contributors. Use of this source code is governed by the Apache 2.0 license that can be found in the LICENSE file.
+// Copyright 2000-2024 JetBrains s.r.o. and contributors. Use of this source code is governed by the Apache 2.0 license.
 package com.intellij.util.lang;

 import org.jetbrains.annotations.ApiStatus;
 import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;

 import java.io.EOFException;
 import java.io.IOException;
 import java.io.InputStream;
 import java.nio.ByteBuffer;
+import java.util.function.IntFunction;
 import java.util.zip.DataFormatException;
 import java.util.zip.Inflater;
 import java.util.zip.ZipException;
@@ -111,14 +113,14 @@ public final class ImmutableZipEntry {

  @ApiStatus.Internal
  public InputStream getInputStream(@NotNull HashMapZipFile file) throws IOException {
-    return new DirectByteBufferBackedInputStream(getByteBuffer(file), method == DEFLATED);
+    return new DirectByteBufferBackedInputStream(getByteBuffer(file, null), method == DEFLATED);
  }

  /**
   * Release returned buffer using {@link ZipFile#releaseBuffer} after use.
   */
  @ApiStatus.Internal
-  public ByteBuffer getByteBuffer(@NotNull HashMapZipFile file) throws IOException {
+  public ByteBuffer getByteBuffer(@NotNull HashMapZipFile file, @Nullable IntFunction<ByteBuffer> allocator) throws IOException {
    if (uncompressedSize < 0) {
      throw new IOException("no data");
    }
@@ -135,7 +137,9 @@ public final class ImmutableZipEntry {
        Inflater inflater = new Inflater(true);
        inflater.setInput(inputBuffer);
        try {
-          ByteBuffer result = DirectByteBufferPool.DEFAULT_POOL.allocate(uncompressedSize);
+          ByteBuffer result = allocator == null
+                              ? DirectByteBufferPool.DEFAULT_POOL.allocate(uncompressedSize)
+                              : allocator.apply(uncompressedSize);
          while (result.hasRemaining()) {
            if (inflater.inflate(result) == 0) {
              throw new IllegalStateException("Inflater wants input, but input was already set");