Repository: 7mind/izumi
Pull request #1926 (2023-04-06 15:05:39)

Separate abstraction for log buffer with IoC

codecov-commenter: ## [Codecov](https://codecov.io/gh/7mind/izumi/pull/1926?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=7mind) Report
> Merging [#1926](https://codecov.io/gh/7mind/izumi/pull/1926?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=7mind) (d816ca2) into [develop](https://codecov.io/gh/7mind/izumi/commit/4ea3902e5b9a1e8a74a9ef2431f66ccfaffb6f3c?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=7mind) (4ea3902) will **increase** coverage by `0.03%`.
> The diff coverage is `93.33%`.

> :exclamation: Current head d816ca2 differs from pull request most recent head a472152. Consider uploading reports for the commit a472152 to get more accurate results

:mega: This organization is not using Codecov’s [GitHub App Integration](https://github.com/apps/codecov). We recommend you install it so Codecov can continue to function properly for your repositories. [Learn more](https://about.codecov.io/blog/codecov-is-updating-its-github-integration/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=7mind)

```diff
@@             Coverage Diff             @@
##           develop    #1926      +/-   ##
===========================================
+ Coverage    64.54%   64.57%   +0.03%     
===========================================
  Files          542      542              
  Lines        10923    10920       -3     
===========================================
+ Hits          7050     7052       +2     
+ Misses        3873     3868       -5     
```


Patch:
diff --git a/distage/distage-framework/.jvm/src/main/scala/izumi/distage/roles/RoleAppBootLoggerModule.scala b/distage/distage-framework/.jvm/src/main/scala/izumi/distage/roles/RoleAppBootLoggerModule.scala
index 3e9c7f7e1..3f3083865 100644
--- a/distage/distage-framework/.jvm/src/main/scala/izumi/distage/roles/RoleAppBootLoggerModule.scala
+++ b/distage/distage-framework/.jvm/src/main/scala/izumi/distage/roles/RoleAppBootLoggerModule.scala
@@ -4,9 +4,10 @@ import izumi.distage.config.model.AppConfig
 import izumi.distage.model.definition.ModuleDef
 import izumi.distage.modules.DefaultModule
 import izumi.distage.roles.launcher.*
-import izumi.logstage.api.logger.LogRouter
+import izumi.logstage.api.logger.{LogQueue, LogRouter}
 import izumi.logstage.api.{IzLogger, Log}
 import izumi.reflect.TagK
+import logstage.ThreadingLogQueue
 
 class RoleAppBootLoggerModule[F[_]: TagK: DefaultModule]() extends ModuleDef {
   make[EarlyLoggerFactory].from[EarlyLoggerFactory.EarlyLoggerFactoryImpl]
@@ -14,6 +15,7 @@ class RoleAppBootLoggerModule[F[_]: TagK: DefaultModule]() extends ModuleDef {
   make[LogConfigLoader].from[LogConfigLoader.LogConfigLoaderImpl]
   make[RouterFactory].from[RouterFactory.RouterFactoryImpl]
   make[LateLoggerFactory].from[LateLoggerFactory.LateLoggerFactoryImpl]
+  make[LogQueue].fromResource(ThreadingLogQueue.resource())
 
   make[Log.Level].named("early").fromValue(Log.Level.Info)
   make[IzLogger].named("early").from {
diff --git a/distage/distage-framework/.jvm/src/main/scala/izumi/distage/roles/launcher/LateLoggerFactory.scala b/distage/distage-framework/.jvm/src/main/scala/izumi/distage/roles/launcher/LateLoggerFactory.scala
index 0602999a8..f6787e894 100644
--- a/distage/distage-framework/.jvm/src/main/scala/izumi/distage/roles/launcher/LateLoggerFactory.scala
+++ b/distage/distage-framework/.jvm/src/main/scala/izumi/distage/roles/launcher/LateLoggerFactory.scala
@@ -4,9 +4,8 @@ import distage.Lifecycle
 import izumi.distage.roles.launcher.LogConfigLoader.DeclarativeLoggerConfig
 import izumi.fundamentals.platform.functional.Identity
 import izumi.logstage.adapter.jul.LogstageJulLogger
-import izumi.logstage.api.logger.LogRouter
+import izumi.logstage.api.logger.{LogQueue, LogRouter}
 import izumi.logstage.api.routing.StaticLogRouter
-import logstage.{ConfigurableLogRouter, QueueingSink}
 
 trait LateLoggerFactory {
   def makeLateLogRouter(config: DeclarativeLoggerConfig): Lifecycle[Identity, LogRouter]
@@ -14,11 +13,12 @@ trait LateLoggerFactory {
 
 object LateLoggerFactory {
   class LateLoggerFactoryImpl(
-    routerFactory: RouterFactory
+    routerFactory: RouterFactory,
+    buffer: LogQueue,
   ) extends LateLoggerFactory {
     def makeLateLogRouter(config: DeclarativeLoggerConfig): Lifecycle[Identity, LogRouter] = {
       for {
-        router <- createThreadingRouter(config)
+        router <- Lifecycle.pure(routerFactory.createRouter(config, buffer))
         _ <- Lifecycle.make[Identity, Unit](StaticLogRouter.instance.setup(router))(_ => ())
         _ <- Lifecycle
           .make[Identity, Option[AutoCloseable]] {
@@ -34,17 +34,5 @@ object LateLoggerFactory {
         router
       }
     }
-
-    protected final def createThreadingRouter(config: DeclarativeLoggerConfig): Lifecycle[Identity, ConfigurableLogRouter] = {
-      Lifecycle
-        .make[Identity, (ConfigurableLogRouter, QueueingSink)] {
-          val (router, queueingSink) = routerFactory.createRouter(config)(sink => new QueueingSink(sink))
-          queueingSink.start()
-          (router, queueingSink)
-        } {
-          case (_, sink) => sink.close()
-        }.map(_._1)
-    }
-
   }
 }
diff --git a/distage/distage-framework/.jvm/src/main/scala/izumi/distage/roles/launcher/RouterFactory.scala b/distage/distage-framework/.jvm/src/main/scala/izumi/distage/roles/launcher/RouterFactory.scala
index e41ca89cf..d6e6293e4 100644
--- a/distage/distage-framework/.jvm/src/main/scala/izumi/distage/roles/launcher/RouterFactory.scala
+++ b/distage/distage-framework/.jvm/src/main/scala/izumi/distage/roles/launcher/RouterFactory.scala
@@ -2,26 +2,26 @@ package izumi.distage.roles.launcher
 
 import izumi.distage.roles.launcher.LogConfigLoader.{DeclarativeLoggerConfig, LoggerFormat}
 import izumi.logstage.api.config.{LoggerConfig, LoggerPathConfig}
-import izumi.logstage.api.logger.LogSink
+import izumi.logstage.api.logger.LogQueue
 import izumi.logstage.api.rendering.StringRenderingPolicy
 import izumi.logstage.api.routing.LogConfigServiceImpl
 import logstage.circe.LogstageCirceRenderingPolicy
 import logstage.{ConfigurableLogRouter, ConsoleSink}
 
 trait RouterFactory {
-  def createRouter[A <: LogSink](config: DeclarativeLoggerConfig)(wrap: ConsoleSink => A): (ConfigurableLogRouter, A)
+  def createRouter(config: DeclarativeLoggerConfig, buffer: LogQueue): ConfigurableLogRouter
 }
 
 object RouterFactory {
   class RouterFactoryImpl extends RouterFactory {
-    override def createRouter[A <: LogSink](config: DeclarativeLoggerConfig)(wrap: ConsoleSink => A): (ConfigurableLogRouter, A) = {
+    override def createRouter(config: DeclarativeLoggerConfig, buffer: LogQueue): ConfigurableLogRouter = {
       val policy = config.format match {
         case LoggerFormat.Json =>
           new LogstageCirceRenderingPolicy()
         case LoggerFormat.Text =>
           new StringRenderingPolicy(config.rendering, None)
       }
-      val sink = wrap(new ConsoleSink(policy))
+      val sink = new ConsoleSink(policy)
       val sinks = List(sink)
       val levelConfigs = config.levels.map {
         case (pkg, level) =>
@@ -32,10 +32,11 @@ object RouterFactory {
       val router = new ConfigurableLogRouter(
         new LogConfigServiceImpl(
           LoggerConfig(LoggerPathConfig(config.rootLevel, sinks), levelConfigs)
-        )
+        ),
+        buffer,
       )
 
-      (router, sink)
+      router
     }
   }
 }
diff --git a/logstage/logstage-core/.js/src/main/scala/izumi/logstage/api/routing/LogConfigServiceImpl.scala b/logstage/logstage-core/.js/src/main/scala/izumi/logstage/api/routing/LogConfigServiceImpl.scala
index c5c3e4da1..9b9931c97 100644
--- a/logstage/logstage-core/.js/src/main/scala/izumi/logstage/api/routing/LogConfigServiceImpl.scala
+++ b/logstage/logstage-core/.js/src/main/scala/izumi/logstage/api/routing/LogConfigServiceImpl.scala
@@ -1,42 +1,41 @@
-package izumi.logstage.api.routing
-
-import izumi.logstage.api.Log
-import izumi.logstage.api.config.{LogConfigService, LogEntryConfig, LoggerConfig, LoggerPathConfig}
-
-import scala.collection.mutable
-
-class LogConfigServiceImpl(loggerConfig: LoggerConfig) extends LogConfigService {
-  override def threshold(e: Log.LoggerId): Log.Level = {
-    configFor(e).threshold
-  }
-
-  override def config(e: Log.Entry): LogEntryConfig = {
-    LogEntryConfig(configFor(e.context.static.id).sinks)
-  }
-
-  @inline private[this] def configFor(e: Log.LoggerId): LoggerPathConfig = {
-    configCache.getOrElseUpdate(e.id, findConfig(e.id))
-  }
-
-  // this should be efficient but may take some memory. Most likely we should use prefix tree here
-  private[this] val configCache = new mutable.HashMap[String, LoggerPathConfig]()
-
-  private[this] def findConfig(id: String): LoggerPathConfig = {
-    val parts = id.split('.')
-
-    // this generates a list of all the prefixes, right to left (com.mycompany.lib.Class, com.mycompany.lib, ...)
-    Iterator
-      .iterate(parts)(_.init)
-      .take(parts.length)
-      .map(_.mkString("."))
-      .map(id => loggerConfig.entries.get(id))
-      .find(_.nonEmpty)
-      .flatten
-      .getOrElse(loggerConfig.root)
-  }
-
-  override def close(): Unit = {
-    (loggerConfig.root.sinks ++ loggerConfig.entries.values.flatMap(_.sinks)).foreach(_.close())
-    configCache.clear()
-  }
-}
+//package izumi.logstage.api.routing
+//
+//import izumi.logstage.api.Log
+//import izumi.logstage.api.config.{LogConfigService, LogEntryConfig, LoggerConfig, LoggerPathConfig}
+//
+//import scala.collection.mutable
+//
+//class LogConfigServiceImpl(loggerConfig: LoggerConfig) extends LogConfigService with AutoCloseable {
+//  override def threshold(e: Log.LoggerId): Log.Level = {
+//    configFor(e).threshold
+//  }
+//
+//  override def config(e: Log.Entry): LogEntryConfig = {
+//    LogEntryConfig(configFor(e.context.static.id).sinks)
+//  }
+//
+//  @inline private[this] def configFor(e: Log.LoggerId): LoggerPathConfig = {
+//    configCache.getOrElseUpdate(e.id, findConfig(e.id))
+//  }
+//
+//  // this should be efficient but may take some memory. Most likely we should use prefix tree here
+//  private[this] val configCache = new mutable.HashMap[String, LoggerPathConfig]()
+//
+//  private[this] def findConfig(id: String): LoggerPathConfig = {
+//    val parts = id.split('.')
+//
+//    // this generates a list of all the prefixes, right to left (com.mycompany.lib.Class, com.mycompany.lib, ...)
+//    Iterator
+//      .iterate(parts)(_.init)
+//      .take(parts.length)
+//      .map(_.mkString("."))
+//      .map(id => loggerConfig.entries.get(id))
+//      .find(_.nonEmpty)
+//      .flatten
+//      .getOrElse(loggerConfig.root)
+//  }
+//
+//  override def close(): Unit = {
+//    configCache.clear()
+//  }
+//}
diff --git a/logstage/logstage-core/.js/src/main/scala/izumi/logstage/sink/QueueingSink.scala b/logstage/logstage-core/.js/src/main/scala/izumi/logstage/sink/QueueingSink.scala
deleted file mode 100644
index 55037bcde..000000000
--- a/logstage/logstage-core/.js/src/main/scala/izumi/logstage/sink/QueueingSink.scala
+++ /dev/null
@@ -1,45 +0,0 @@
-package izumi.logstage.sink
-
-import izumi.fundamentals.platform.language.Quirks
-import izumi.logstage.api.Log
-import izumi.logstage.api.logger.LogSink
-
-import scala.concurrent.duration._
-
-class QueueingSink(target: LogSink, sleepTime: FiniteDuration = 50.millis) extends LogSink with AutoCloseable {
-
-  Quirks.discard(sleepTime)
-
-  def start(): Unit = {}
-
-  def flush(e: Log.Entry): Unit = {
-    target.flush(e)
-  }
-
-  override def close(): Unit = {}
-
-}
-
-object QueueingSink {
-
-  trait Step {
-    def continue: Boolean
-
-    def onStep(): Unit
-  }
-
-  class CountingStep(max: Int) extends Step {
-    var counter: Int = 0
-
-    override def continue: Boolean = counter <= max
-
-    override def onStep(): Unit = counter += 1
-  }
-
-  object NullStep extends Step {
-    override def continue: Boolean = true
-
-    override def onStep(): Unit = {}
-  }
-
-}
diff --git a/logstage/logstage-core/.js/src/main/scala/izumi/logstage/sink/ThreadingLogQueue.scala b/logstage/logstage-core/.js/src/main/scala/izumi/logstage/sink/ThreadingLogQueue.scala
new file mode 100644
index 000000000..26087102e
--- /dev/null
+++ b/logstage/logstage-core/.js/src/main/scala/izumi/logstage/sink/ThreadingLogQueue.scala
@@ -0,0 +1,20 @@
+package izumi.logstage.sink
+
+import izumi.logstage.api.Log
+import izumi.logstage.api.logger.{LogQueue, LogSink}
+import izumi.fundamentals.platform.language.Quirks.*
+
+import scala.concurrent.duration.FiniteDuration
+
+class ThreadingLogQueue(sleepTime: FiniteDuration, batchSize: Int) extends LogQueue with AutoCloseable {
+  (sleepTime, batchSize).discard()
+  def start(): Unit = {}
+
+  override def append(entry: Log.Entry, target: LogSink): Unit = {
+    target.flush(entry)
+  }
+
+  override def close(): Unit = {}
+}
+
+object ThreadingLogQueue {}
diff --git a/logstage/logstage-core/.jvm/src/main/scala/izumi/logstage/sink/QueueingSink.scala b/logstage/logstage-core/.jvm/src/main/scala/izumi/logstage/sink/ThreadingLogQueue.scala
similarity index 73%
rename from logstage/logstage-core/.jvm/src/main/scala/izumi/logstage/sink/QueueingSink.scala
rename to logstage/logstage-core/.jvm/src/main/scala/izumi/logstage/sink/ThreadingLogQueue.scala
index 5a3b2b041..a0de1dd84 100644
--- a/logstage/logstage-core/.jvm/src/main/scala/izumi/logstage/sink/QueueingSink.scala
+++ b/logstage/logstage-core/.jvm/src/main/scala/izumi/logstage/sink/ThreadingLogQueue.scala
@@ -1,20 +1,34 @@
 package izumi.logstage.sink
 
+import izumi.functional.lifecycle.Lifecycle
 import izumi.fundamentals.platform.IzumiProject
 import izumi.fundamentals.platform.console.TrivialLogger
 import izumi.fundamentals.platform.console.TrivialLogger.Config
+import izumi.fundamentals.platform.functional.Identity
 import izumi.fundamentals.platform.language.Quirks.*
 import izumi.logstage.DebugProperties
 import izumi.logstage.api.Log
-import izumi.logstage.api.logger.LogSink
+import izumi.logstage.api.logger.{LogQueue, LogSink}
 
 import java.util.concurrent.ConcurrentLinkedQueue
 import java.util.concurrent.atomic.AtomicBoolean
 import scala.concurrent.duration.*
 
-class QueueingSink(target: LogSink, sleepTime: FiniteDuration = 50.millis) extends LogSink {
-  private val queue = new ConcurrentLinkedQueue[Log.Entry]()
-  private val maxBatchSize = 100
+object ThreadingLogQueue {
+  case class LoggingAction(entry: Log.Entry, target: LogSink)
+
+  def resource(sleepTime: FiniteDuration = 50.millis, batchSize: Int = 100): Lifecycle[Identity, ThreadingLogQueue] = Lifecycle
+    .make[Identity, ThreadingLogQueue] {
+      val buffer = new ThreadingLogQueue(sleepTime, batchSize)
+      buffer.start()
+      buffer
+    } {
+      _.close()
+    }
+}
+
+class ThreadingLogQueue(sleepTime: FiniteDuration, batchSize: Int) extends LogQueue with AutoCloseable {
+  private val queue = new ConcurrentLinkedQueue[ThreadingLogQueue.LoggingAction]()
   private val stop = new AtomicBoolean(false)
 
   private val fallback = TrivialLogger.make[FallbackConsoleSink](DebugProperties.`izumi.logstage.routing.log-failures`.name, Config(forceLog = true))
@@ -55,7 +69,7 @@ class QueueingSink(target: LogSink, sleepTime: FiniteDuration = 50.millis) exten
           fallback.log(IzumiProject.bugReportPrompt("LogStage shutdown hook had been interrupted"))
       }
 
-      target.sync()
+      // target.sync()
 
       if (!queue.isEmpty) {
         fallback.log(IzumiProject.bugReportPrompt("LogStage queue wasn't empty and the end of the polling thread"))
@@ -68,7 +82,7 @@ class QueueingSink(target: LogSink, sleepTime: FiniteDuration = 50.millis) exten
       while (!stop.get()) {
         try {
           // in case queue was empty we sleep a bit (it's a sane heuristic), otherwise it's better to continue working asap
-          if (doFlush(maxBatchSize)) {
+          if (doFlush(batchSize)) {
             Thread.`yield`()
           } else {
             Thread.sleep(sleepTime.toMillis)
@@ -92,30 +106,30 @@ class QueueingSink(target: LogSink, sleepTime: FiniteDuration = 50.millis) exten
     pollingThread.start()
   }
 
-  def flush(e: Log.Entry): Unit = {
-    if (!stop.get()) { // Once shutdown is reqested the queue will stop receiving new messages, so it'll become finite
-      queue.add(e).discard()
-    } else {
-      target.flush(e)
-      target.sync()
-    }
-  }
-
   /**
     * @return true if any messages were processed
     */
   private def doFlush(maxBatch: Int): Boolean = {
     var cc: Int = 0
-    var entry: Log.Entry = null
+    var entry: ThreadingLogQueue.LoggingAction = null
 
     while ({
       entry = queue.poll()
       if (entry != null)
-        target.flush(entry)
+        entry.target.flush(entry.entry)
       cc += 1
       cc <= maxBatch && entry != null
     }) {}
 
     entry != null
   }
+
+  override def append(entry: Log.Entry, target: LogSink): Unit = {
+    if (!stop.get()) { // Once shutdown is reqested the queue will stop receiving new messages, so it'll become finite
+      queue.add(ThreadingLogQueue.LoggingAction(entry, target)).discard()
+    } else {
+      target.flush(entry)
+      target.sync()
+    }
+  }
 }
diff --git a/logstage/logstage-core/src/main/scala/izumi/logstage/api/IzLoggerConvenienceApi.scala b/logstage/logstage-core/src/main/scala/izumi/logstage/api/IzLoggerConvenienceApi.scala
index 97b61326e..1fb55119f 100644
--- a/logstage/logstage-core/src/main/scala/izumi/logstage/api/IzLoggerConvenienceApi.scala
+++ b/logstage/logstage-core/src/main/scala/izumi/logstage/api/IzLoggerConvenienceApi.scala
@@ -1,7 +1,7 @@
 package izumi.logstage.api
 
 import izumi.logstage.api.Log.CustomContext
-import izumi.logstage.api.logger.{LogRouter, LogSink, RoutingLogger}
+import izumi.logstage.api.logger.{LogQueue, LogRouter, LogSink, RoutingLogger}
 import izumi.logstage.api.routing.ConfigurableLogRouter
 import izumi.logstage.sink.ConsoleSink
 
@@ -24,16 +24,29 @@ trait IzLoggerConvenienceApi[Logger <: RoutingLogger] {
   /**
     * By default, a basic colored console logger with global [[Level.Trace]] minimum threshold
     */
-  final def apply(threshold: Log.Level = Log.Level.Trace, sink: LogSink = ConsoleSink.ColoredConsoleSink, levels: Map[String, Log.Level] = Map.empty): Logger = {
-    make(ConfigurableLogRouter(threshold, sink, levels))
+  final def apply(
+    threshold: Log.Level = Log.Level.Trace,
+    sink: LogSink = ConsoleSink.ColoredConsoleSink,
+    levels: Map[String, Log.Level] = Map.empty,
+    buffer: LogQueue = LogQueue.Immediate,
+  ): Logger = {
+    make(ConfigurableLogRouter(threshold, sink, levels, buffer))
   }
 
   final def apply(threshold: Log.Level, sinks: Seq[LogSink]): Logger = {
-    make(ConfigurableLogRouter(threshold, sinks))
+    make(ConfigurableLogRouter(threshold, sinks, LogQueue.Immediate))
+  }
+
+  final def apply(threshold: Log.Level, sinks: Seq[LogSink], buffer: LogQueue): Logger = {
+    make(ConfigurableLogRouter(threshold, sinks, buffer))
+  }
+
+  final def apply(threshold: Log.Level, sinks: Seq[LogSink], levels: Map[String, Log.Level], buffer: LogQueue): Logger = {
+    make(ConfigurableLogRouter(threshold, sinks, levels, buffer))
   }
 
   final def apply(threshold: Log.Level, sinks: Seq[LogSink], levels: Map[String, Log.Level]): Logger = {
-    make(ConfigurableLogRouter(threshold, sinks, levels))
+    make(ConfigurableLogRouter(threshold, sinks, levels, LogQueue.Immediate))
   }
 
   final def apply(router: LogRouter): Logger = {
diff --git a/logstage/logstage-core/src/main/scala/izumi/logstage/api/config/LogConfigService.scala b/logstage/logstage-core/src/main/scala/izumi/logstage/api/config/LogConfigService.scala
index bccf74ffa..90724d19a 100644
--- a/logstage/logstage-core/src/main/scala/izumi/logstage/api/config/LogConfigService.scala
+++ b/logstage/logstage-core/src/main/scala/izumi/logstage/api/config/LogConfigService.scala
@@ -2,7 +2,7 @@ package izumi.logstage.api.config
 
 import izumi.logstage.api.Log
 
-trait LogConfigService extends AutoCloseable {
+trait LogConfigService {
   def threshold(e: Log.LoggerId): Log.Level
   def config(e: Log.Entry): LogEntryConfig
 }
diff --git a/logstage/logstage-core/src/main/scala/izumi/logstage/api/logger/LogQueue.scala b/logstage/logstage-core/src/main/scala/izumi/logstage/api/logger/LogQueue.scala
new file mode 100644
index 000000000..05f057541
--- /dev/null
+++ b/logstage/logstage-core/src/main/scala/izumi/logstage/api/logger/LogQueue.scala
@@ -0,0 +1,16 @@
+package izumi.logstage.api.logger
+
+import izumi.logstage.api.Log
+
+trait LogQueue extends AutoCloseable {
+  def append(entry: Log.Entry, target: LogSink): Unit
+  def close(): Unit = ()
+}
+
+object LogQueue {
+  object Immediate extends LogQueue {
+    override def append(entry: Log.Entry, target: LogSink): Unit = {
+      target.flush(entry)
+    }
+  }
+}
diff --git a/logstage/logstage-core/src/main/scala/izumi/logstage/api/logger/LogRouter.scala b/logstage/logstage-core/src/main/scala/izumi/logstage/api/logger/LogRouter.scala
index cbc5ca069..59a9f2f4c 100644
--- a/logstage/logstage-core/src/main/scala/izumi/logstage/api/logger/LogRouter.scala
+++ b/logstage/logstage-core/src/main/scala/izumi/logstage/api/logger/LogRouter.scala
@@ -15,8 +15,13 @@ trait LogRouter extends AutoCloseable {
 }
 
 object LogRouter {
-  def apply(threshold: Log.Level = Log.Level.Trace, sink: LogSink = ConsoleSink.ColoredConsoleSink, levels: Map[String, Log.Level] = Map.empty): ConfigurableLogRouter = {
-    ConfigurableLogRouter(threshold, Seq(sink), levels)
+  def apply(
+             threshold: Log.Level = Log.Level.Trace,
+             sink: LogSink = ConsoleSink.ColoredConsoleSink,
+             levels: Map[String, Log.Level] = Map.empty,
+             buffer: LogQueue = LogQueue.Immediate,
+  ): ConfigurableLogRouter = {
+    ConfigurableLogRouter(threshold, Seq(sink), levels, buffer)
   }
 
   lazy val debugRouter: LogRouter = new LogRouter {
diff --git a/logstage/logstage-core/src/main/scala/izumi/logstage/api/logger/LogSink.scala b/logstage/logstage-core/src/main/scala/izumi/logstage/api/logger/LogSink.scala
index 754973290..c79fc5363 100644
--- a/logstage/logstage-core/src/main/scala/izumi/logstage/api/logger/LogSink.scala
+++ b/logstage/logstage-core/src/main/scala/izumi/logstage/api/logger/LogSink.scala
@@ -2,8 +2,11 @@ package izumi.logstage.api.logger
 
 import izumi.logstage.api.Log
 
-trait LogSink extends AutoCloseable {
+trait LogSink {
   def flush(e: Log.Entry): Unit
   def sync(): Unit = ()
-  def close(): Unit = ()
 }
+
+
+
+
diff --git a/logstage/logstage-core/src/main/scala/izumi/logstage/api/routing/ConfigurableLogRouter.scala b/logstage/logstage-core/src/main/scala/izumi/logstage/api/routing/ConfigurableLogRouter.scala
index 39d80b039..a7d353f08 100644
--- a/logstage/logstage-core/src/main/scala/izumi/logstage/api/routing/ConfigurableLogRouter.scala
+++ b/logstage/logstage-core/src/main/scala/izumi/logstage/api/routing/ConfigurableLogRouter.scala
@@ -5,14 +5,15 @@ import izumi.fundamentals.platform.console.TrivialLogger.Config
 import izumi.logstage.DebugProperties
 import izumi.logstage.api.Log
 import izumi.logstage.api.config.{LogConfigService, LoggerConfig, LoggerPathConfig}
-import izumi.logstage.api.logger.{LogRouter, LogSink}
+import izumi.logstage.api.logger.{LogQueue, LogRouter, LogSink}
 import izumi.logstage.sink.{ConsoleSink, FallbackConsoleSink}
 
 import scala.annotation.nowarn
 import scala.util.control.NonFatal
 
 class ConfigurableLogRouter(
-  logConfigService: LogConfigService
+  logConfigService: LogConfigService,
+  buffer: LogQueue,
 ) extends LogRouter {
   private final val fallback = TrivialLogger.make[FallbackConsoleSink](DebugProperties.`izumi.logstage.routing.log-failures`.name, Config(forceLog = true))
 
@@ -24,7 +25,8 @@ class ConfigurableLogRouter(
     sinks.foreach {
       sink =>
         try {
-          sink.flush(entry)
+          buffer.append(entry, sink)
+//          sink.flush(entry)
         } catch {
           case NonFatal(e) =>
             fallback.log(s"Log sink $sink failed", e)
@@ -32,26 +34,27 @@ class ConfigurableLogRouter(
     }
   }
 
-  override def close(): Unit = {
-    logConfigService.close()
-  }
-
   override def acceptable(id: Log.LoggerId, messageLevel: Log.Level): Boolean = {
     logConfigService.threshold(id) <= messageLevel
   }
 }
 
 object ConfigurableLogRouter {
-  def apply(threshold: Log.Level = Log.Level.Trace, sink: LogSink = ConsoleSink.ColoredConsoleSink, levels: Map[String, Log.Level] = Map.empty): ConfigurableLogRouter = {
-    ConfigurableLogRouter(threshold, Seq(sink), levels)
+  def apply(
+             threshold: Log.Level = Log.Level.Trace,
+             sink: LogSink = ConsoleSink.ColoredConsoleSink,
+             levels: Map[String, Log.Level] = Map.empty,
+             buffer: LogQueue = LogQueue.Immediate,
+  ): ConfigurableLogRouter = {
+    ConfigurableLogRouter(threshold, Seq(sink), levels, buffer)
   }
 
-  def apply(threshold: Log.Level, sinks: Seq[LogSink]): ConfigurableLogRouter = {
-    ConfigurableLogRouter(threshold, sinks, Map.empty[String, Log.Level])
+  def apply(threshold: Log.Level, sinks: Seq[LogSink], buffer: LogQueue): ConfigurableLogRouter = {
+    ConfigurableLogRouter(threshold, sinks, Map.empty[String, Log.Level], buffer)
   }
 
   @nowarn("msg=Unused import")
-  def apply(threshold: Log.Level, sinks: Seq[LogSink], levels: Map[String, Log.Level]): ConfigurableLogRouter = {
+  def apply(threshold: Log.Level, sinks: Seq[LogSink], levels: Map[String, Log.Level], buffer: LogQueue): ConfigurableLogRouter = {
     import scala.collection.compat._
 
     val rootConfig = LoggerPathConfig(threshold, sinks)
@@ -59,8 +62,8 @@ object ConfigurableLogRouter {
 
     val configService = new LogConfigServiceImpl(LoggerConfig(rootConfig, levelConfigs))
 
-    ConfigurableLogRouter(configService)
+    ConfigurableLogRouter(configService, buffer)
   }
 
-  def apply(logConfigService: LogConfigService): ConfigurableLogRouter = new ConfigurableLogRouter(logConfigService)
+  def apply(logConfigService: LogConfigService, buffer: LogQueue): ConfigurableLogRouter = new ConfigurableLogRouter(logConfigService, buffer)
 }
diff --git a/logstage/logstage-core/.jvm/src/main/scala/izumi/logstage/api/routing/LogConfigServiceImpl.scala b/logstage/logstage-core/src/main/scala/izumi/logstage/api/routing/LogConfigServiceImpl.scala
similarity index 96%
rename from logstage/logstage-core/.jvm/src/main/scala/izumi/logstage/api/routing/LogConfigServiceImpl.scala
rename to logstage/logstage-core/src/main/scala/izumi/logstage/api/routing/LogConfigServiceImpl.scala
index 06a1e5ea6..d00a20281 100644
--- a/logstage/logstage-core/.jvm/src/main/scala/izumi/logstage/api/routing/LogConfigServiceImpl.scala
+++ b/logstage/logstage-core/src/main/scala/izumi/logstage/api/routing/LogConfigServiceImpl.scala
@@ -40,9 +40,9 @@ class LogConfigServiceImpl(loggerConfig: LoggerConfig) extends LogConfigService
     }
   }
 
-  override def close(): Unit = {
-    (loggerConfig.root.sinks ++ loggerConfig.entries.values.flatMap(_.sinks)).foreach(_.close())
-  }
+//  override def close(): Unit = {
+//    (loggerConfig.root.sinks ++ loggerConfig.entries.values.flatMap(_.sinks)).foreach(_.close())
+//  }
 
   private[this] def print(node: LogTreeNode, level: Int): String = {
     val sub = node.sub.values.map(s => print(s, level + 1))
diff --git a/logstage/logstage-core/src/main/scala/logstage/LogStage.scala b/logstage/logstage-core/src/main/scala/logstage/LogStage.scala
index d76eb1d8e..dfb159322 100644
--- a/logstage/logstage-core/src/main/scala/logstage/LogStage.scala
+++ b/logstage/logstage-core/src/main/scala/logstage/LogStage.scala
@@ -12,8 +12,11 @@ trait LogStage {
 
   type LogIORaw[F[_], E <: AnyEncoded] = izumi.logstage.api.logger.LogIORaw[F, E]
 
-  type QueueingSink = sink.QueueingSink
-  val QueueingSink: sink.QueueingSink.type = sink.QueueingSink
+  type LogQueue = izumi.logstage.api.logger.LogQueue
+  val LogQueue: izumi.logstage.api.logger.LogQueue.type = izumi.logstage.api.logger.LogQueue
+
+  type ThreadingLogQueue = sink.ThreadingLogQueue
+  val ThreadingLogQueue: sink.ThreadingLogQueue.type = sink.ThreadingLogQueue
 
   type ConfigurableLogRouter = api.routing.ConfigurableLogRouter
   val ConfigurableLogRouter: api.routing.ConfigurableLogRouter.type = api.routing.ConfigurableLogRouter
diff --git a/logstage/logstage-core/src/main/scala/logstage/package.scala b/logstage/logstage-core/src/main/scala/logstage/package.scala
index 8f644f9e8..d605967af 100644
--- a/logstage/logstage-core/src/main/scala/logstage/package.scala
+++ b/logstage/logstage-core/src/main/scala/logstage/package.scala
@@ -23,8 +23,11 @@ package object logstage extends LogStage {
 
   override type LogIORaw[F[_], E <: AnyEncoded] = izumi.logstage.api.logger.LogIORaw[F, E]
 
-  override type QueueingSink = sink.QueueingSink
-  override final val QueueingSink: sink.QueueingSink.type = sink.QueueingSink
+  override type LogQueue = izumi.logstage.api.logger.LogQueue
+  override final val LogQueue: izumi.logstage.api.logger.LogQueue.type = izumi.logstage.api.logger.LogQueue
+
+  override type ThreadingLogQueue = sink.ThreadingLogQueue
+  override final val ThreadingLogQueue: sink.ThreadingLogQueue.type = sink.ThreadingLogQueue
 
   override type ConfigurableLogRouter = api.routing.ConfigurableLogRouter
   override final val ConfigurableLogRouter: api.routing.ConfigurableLogRouter.type = api.routing.ConfigurableLogRouter


Test patch:
diff --git a/distage/distage-framework/.jvm/src/test/scala/izumi/distage/roles/test/RoleAppTest.scala b/distage/distage-framework/.jvm/src/test/scala/izumi/distage/roles/test/RoleAppTest.scala
index ff01455e2..659fae08d 100644
--- a/distage/distage-framework/.jvm/src/test/scala/izumi/distage/roles/test/RoleAppTest.scala
+++ b/distage/distage-framework/.jvm/src/test/scala/izumi/distage/roles/test/RoleAppTest.scala
@@ -70,8 +70,9 @@ class RoleAppTest extends AnyWordSpec with WithProperties {
         )
       )
 
-      assert(probe.resources.getStartedCloseables() == probe.resources.getClosedCloseables().reverse.filter(!_.isInstanceOf[LogSink]))
-      assert(probe.resources.getStartedCloseables() != probe.resources.getClosedCloseables().reverse)
+//      assert(probe.resources.getStartedCloseables() == probe.resources.getClosedCloseables().reverse.filter(!_.isInstanceOf[LogSink]))
+      assert(probe.resources.getStartedCloseables() == probe.resources.getClosedCloseables().reverse)
+
       assert(
         probe.resources.getCheckedResources().toSet == Set[IntegrationCheck[IO]](probe.locator.get[IntegrationResource0[IO]], probe.locator.get[IntegrationResource1[IO]])
       )
diff --git a/distage/distage-framework/.jvm/src/test/scala/izumi/distage/roles/test/fixtures/AdaptedAutocloseablesCasePlugin.scala b/distage/distage-framework/.jvm/src/test/scala/izumi/distage/roles/test/fixtures/AdaptedAutocloseablesCasePlugin.scala
index 9b13890c4..e86ef11e4 100644
--- a/distage/distage-framework/.jvm/src/test/scala/izumi/distage/roles/test/fixtures/AdaptedAutocloseablesCasePlugin.scala
+++ b/distage/distage-framework/.jvm/src/test/scala/izumi/distage/roles/test/fixtures/AdaptedAutocloseablesCasePlugin.scala
@@ -5,7 +5,6 @@ import izumi.distage.model.definition.Lifecycle
 import izumi.distage.plugins.PluginDef
 import izumi.distage.roles.model.definition.RoleModuleDef
 import izumi.distage.roles.model.{RoleDescriptor, RoleService}
-import izumi.distage.roles.test.fixtures.Fixture.XXX_ResourceEffectsRecorder
 import izumi.fundamentals.platform.cli.model.raw.RawEntrypointParams
 import izumi.logstage.api.Log
 import izumi.logstage.api.logger.LogSink
@@ -15,18 +14,16 @@ class AdaptedAutocloseablesCasePlugin extends PluginDef with RoleModuleDef {
   many[LogSink].add[BrokenSink]
   many[LogSink].add[BrokenSink2]
   many[LogSink].add {
-    new BrokenSink(_)
+    new BrokenSink()
   }
 }
 
-class BrokenSink2(recorder: XXX_ResourceEffectsRecorder[IO]) extends LogSink {
+class BrokenSink2(/*recorder: XXX_ResourceEffectsRecorder[IO]*/ ) extends LogSink {
   override def flush(e: Log.Entry): Unit = {}
-  override def close(): Unit = recorder.onClose(this)
 }
 
-class BrokenSink(recorder: XXX_ResourceEffectsRecorder[IO]) extends LogSink {
+class BrokenSink(/*recorder: XXX_ResourceEffectsRecorder[IO]*/ ) extends LogSink {
   override def flush(e: Log.Entry): Unit = {}
-  override def close(): Unit = recorder.onClose(this)
 }
 
 class AdaptedAutocloseablesCase(
diff --git a/distage/distage-testkit-core/src/main/scala/izumi/distage/testkit/runner/TestkitRunnerModule.scala b/distage/distage-testkit-core/src/main/scala/izumi/distage/testkit/runner/TestkitRunnerModule.scala
index a75650839..3e9a6870e 100644
--- a/distage/distage-testkit-core/src/main/scala/izumi/distage/testkit/runner/TestkitRunnerModule.scala
+++ b/distage/distage-testkit-core/src/main/scala/izumi/distage/testkit/runner/TestkitRunnerModule.scala
@@ -11,6 +11,8 @@ import izumi.distage.testkit.runner.impl.services.TimedAction.TimedActionImpl
 import izumi.distage.testkit.runner.impl.services.TimedActionF.TimedActionFImpl
 import izumi.distage.testkit.runner.impl.{DistageTestRunner, TestPlanner, TestTreeBuilder}
 import izumi.fundamentals.platform.functional.Identity
+import izumi.logstage.api.logger.LogQueue
+import logstage.ThreadingLogQueue
 
 class TestkitRunnerModule[F[_]: TagK: DefaultModule](reporter: TestReporter, isCancellation: Throwable => Boolean) extends ModuleDef {
   addImplicit[TagK[F]]
@@ -30,6 +32,7 @@ class TestkitRunnerModule[F[_]: TagK: DefaultModule](reporter: TestReporter, isC
   make[ExtParTraverse[Identity]].from[ExtParTraverse.ExtParTraverseImpl[Identity]]
 
   make[DistageTestRunner[F]].from[DistageTestRunner[F]]
+  make[LogQueue].fromResource(ThreadingLogQueue.resource())
 }
 
 object TestkitRunnerModule {
diff --git a/distage/distage-testkit-core/src/main/scala/izumi/distage/testkit/runner/impl/TestPlanner.scala b/distage/distage-testkit-core/src/main/scala/izumi/distage/testkit/runner/impl/TestPlanner.scala
index 9bade6ad1..3780aa5d7 100644
--- a/distage/distage-testkit-core/src/main/scala/izumi/distage/testkit/runner/impl/TestPlanner.scala
+++ b/distage/distage-testkit-core/src/main/scala/izumi/distage/testkit/runner/impl/TestPlanner.scala
@@ -21,7 +21,7 @@ import izumi.functional.quasi.{QuasiAsync, QuasiIORunner}
 import izumi.fundamentals.platform.cli.model.raw.RawAppArgs
 import izumi.fundamentals.platform.functional.Identity
 import izumi.logstage.api.IzLogger
-import izumi.logstage.api.logger.LogRouter
+import izumi.logstage.api.logger.{LogQueue, LogRouter}
 
 import scala.annotation.nowarn
 import scala.util.Try
@@ -74,6 +74,7 @@ class TestPlanner[F[_]: TagK: DefaultModule](
   configLoader: TestConfigLoader,
   testTreeBuilder: TestTreeBuilder[F],
   testRunnerLocator: LocatorRef,
+  logBuffer: LogQueue,
 ) {
   // first we need to plan runtime for our monad, which is retained by TestTreeRunner. Identity is also supported.
   private val runtimeGcRoots: Set[DIKey] = Set(
@@ -109,7 +110,7 @@ class TestPlanner[F[_]: TagK: DefaultModule](
                 // test loggers will not create polling threads and will log immediately
                 val logConfigLoader = new LogConfigLoaderImpl(CLILoggerOptions(envExec.logLevel, json = false), configLoadLogger)
                 val logConfig = logConfigLoader.loadLoggingConfig(config)
-                val (router, _) = new RouterFactory.RouterFactoryImpl().createRouter(logConfig)(identity)
+                val router = new RouterFactory.RouterFactoryImpl().createRouter(logConfig, logBuffer)
 
                 prepareGroupPlans(envExec, config, env, tests, router).left.map(bad => (tests, bad))
             }
diff --git a/logstage/logstage-core/.js/src/test/scala/izumi/logstage/sink/LoggingAsyncSinkTest.scala b/logstage/logstage-core/.js/src/test/scala/izumi/logstage/sink/LoggingAsyncSinkTest.scala
index 549f07279..d3a137b8e 100644
--- a/logstage/logstage-core/.js/src/test/scala/izumi/logstage/sink/LoggingAsyncSinkTest.scala
+++ b/logstage/logstage-core/.js/src/test/scala/izumi/logstage/sink/LoggingAsyncSinkTest.scala
@@ -3,14 +3,16 @@ package izumi.logstage.sink
 import izumi.logstage.api.{IzLogger, TestSink}
 import org.scalatest.wordspec.AnyWordSpec
 
+import scala.concurrent.duration.DurationInt
+
 class LoggingAsyncSinkTest extends AnyWordSpec {
   "Log macro" should {
-    "support async sink api in sync mode" in {
+    "support async sinks" in {
       val testSink = new TestSink()
-      val asyncConsoleSinkJson = new QueueingSink(testSink)
+      val asyncConsoleSinkJson = new ThreadingLogQueue(1.millis, 10)
       try {
-        new ExampleService(IzLogger(IzLogger.Level.Trace, asyncConsoleSinkJson)).triggerManyMessages()
-        assert(testSink.fetch().size == 100)
+        new ExampleService(IzLogger(IzLogger.Level.Trace, testSink, buffer = asyncConsoleSinkJson)).triggerManyMessages()
+        //assert(testSink.fetch().isEmpty) // js version is immediate
         asyncConsoleSinkJson.start()
       } finally {
         asyncConsoleSinkJson.close()
diff --git a/logstage/logstage-core/.jvm/src/test/scala/izumi/logstage/sink/LoggingAsyncSinkTest.scala b/logstage/logstage-core/.jvm/src/test/scala/izumi/logstage/sink/LoggingAsyncSinkTest.scala
index 21b30989e..521a461a9 100644
--- a/logstage/logstage-core/.jvm/src/test/scala/izumi/logstage/sink/LoggingAsyncSinkTest.scala
+++ b/logstage/logstage-core/.jvm/src/test/scala/izumi/logstage/sink/LoggingAsyncSinkTest.scala
@@ -3,13 +3,15 @@ package izumi.logstage.sink
 import izumi.logstage.api.{IzLogger, TestSink}
 import org.scalatest.wordspec.AnyWordSpec
 
+import scala.concurrent.duration.DurationInt
+
 class LoggingAsyncSinkTest extends AnyWordSpec {
   "Log macro" should {
     "support async sinks" in {
       val testSink = new TestSink()
-      val asyncConsoleSinkJson = new QueueingSink(testSink)
+      val asyncConsoleSinkJson = new ThreadingLogQueue(1.millis, 10)
       try {
-        new ExampleService(IzLogger(IzLogger.Level.Trace, asyncConsoleSinkJson)).triggerManyMessages()
+        new ExampleService(IzLogger(IzLogger.Level.Trace, testSink, buffer = asyncConsoleSinkJson)).triggerManyMessages()
         assert(testSink.fetch().isEmpty)
         asyncConsoleSinkJson.start()
       } finally {
diff --git a/logstage/logstage-core/src/test/scala/izumi/logstage/api/TestSink.scala b/logstage/logstage-core/src/test/scala/izumi/logstage/api/TestSink.scala
index 0d751277c..50e1c8383 100644
--- a/logstage/logstage-core/src/test/scala/izumi/logstage/api/TestSink.scala
+++ b/logstage/logstage-core/src/test/scala/izumi/logstage/api/TestSink.scala
@@ -3,7 +3,7 @@ package izumi.logstage.api
 import java.util.concurrent.ConcurrentLinkedQueue
 
 import izumi.fundamentals.platform.build.ExposedTestScope
-import izumi.fundamentals.platform.language.Quirks._
+  import izumi.fundamentals.platform.language.Quirks._
 import izumi.logstage.api.logger.LogSink
 import izumi.logstage.api.rendering.RenderingPolicy
 
diff --git a/logstage/logstage-core/src/test/scala/logstage/LogZIOSpec.scala b/logstage/logstage-core/src/test/scala/logstage/LogZIOSpec.scala
index 4cd649ff7..0bf6fcb14 100644
--- a/logstage/logstage-core/src/test/scala/logstage/LogZIOSpec.scala
+++ b/logstage/logstage-core/src/test/scala/logstage/LogZIOSpec.scala
@@ -138,7 +138,7 @@ class LogZIOSpec extends AnyWordSpec {
   private def withTestSink[U](thunk: RIO[LogZIO, U]): TestSink = {
     val sink = new TestSink()
     val logger = LogIO.fromLogger[UIO](
-      IzLogger(Log.Level.Info, List(sink))
+      IzLogger(Log.Level.Info, sink)
     )
     runtime.unsafeRun {
       thunk.provide(Has(logger))
diff --git a/logstage/logstage-rendering-circe/src/test/scala/izumi/logstage/api/rendering/json/LogstageCirceRenderingTest.scala b/logstage/logstage-rendering-circe/src/test/scala/izumi/logstage/api/rendering/json/LogstageCirceRenderingTest.scala
index 511cceb18..416f69487 100644
--- a/logstage/logstage-rendering-circe/src/test/scala/izumi/logstage/api/rendering/json/LogstageCirceRenderingTest.scala
+++ b/logstage/logstage-rendering-circe/src/test/scala/izumi/logstage/api/rendering/json/LogstageCirceRenderingTest.scala
@@ -1,8 +1,9 @@
 package izumi.logstage.api.rendering.json
 
-import io.circe.literal._
-import io.circe.parser._
+import io.circe.literal.*
+import io.circe.parser.*
 import io.circe.{Codec, Json}
+import izumi.logstage.api.logger.LogQueue
 import izumi.logstage.api.rendering.LogstageCodec
 import izumi.logstage.api.strict.IzStrictLogger
 import izumi.logstage.api.{IzLogger, TestSink}
@@ -105,7 +106,7 @@ class LogstageCirceRenderingTest extends AnyWordSpec {
       Seq(testSink)
     }
 
-    (IzLogger(IzLogger.Level.Trace, sinks), testSink)
+    (IzLogger(IzLogger.Level.Trace, sinks, buffer = LogQueue.Immediate), testSink)
   }
 
   def setupJsonStrictLogger(debug: Boolean): (IzStrictLogger, TestSink) = {
@@ -118,7 +119,7 @@ class LogstageCirceRenderingTest extends AnyWordSpec {
       Seq(testSink)
     }
 
-    (IzStrictLogger(IzLogger.Level.Trace, sinks), testSink)
+    (IzStrictLogger(IzLogger.Level.Trace, sinks, buffer = LogQueue.Immediate), testSink)
   }
 }