что я сделал не так

не хватает чтобы кто-то апнул версию тупед схемы...

Наверное тут ты ни при чем. Есть какой-то старый говнокод

val (sync, observable) =
    Observable.multicast[A](MulticastStrategy.publish, OverflowStrategy.DropNew(config.maxBufferSize))
def enqueue(entry: A): Unit = Task.delay(sync.onNext(entry)).startAndForget.runSyncUnsafe(1.minute)

И судя по стеку, все треды сидят вот так кайфуют:

"main-106" #106 daemon prio=5 os_prio=0 cpu=1070.12ms elapsed=100.89s tid=0x00007fc5f8004800 nid=0x28f4 waiting on condition  [0x00007fc5dc9cd000]
   java.lang.Thread.State: TIMED_WAITING (parking)
  at jdk.internal.misc.Unsafe.park(java.base@13.0.1/Native Method)
  - parking to wait for  <0x00000006fa70c900> (a monix.eval.internal.TaskRunSyncUnsafe$OneShotLatch)
  at java.util.concurrent.locks.LockSupport.parkNanos(java.base@13.0.1/LockSupport.java:235)
  at java.util.concurrent.locks.AbstractQueuedSynchronizer.doAcquireSharedNanos(java.base@13.0.1/AbstractQueuedSynchronizer.java:1079)
  at java.util.concurrent.locks.AbstractQueuedSynchronizer.tryAcquireSharedNanos(java.base@13.0.1/AbstractQueuedSynchronizer.java:1369)
  at monix.eval.internal.TaskRunSyncUnsafe$.$anonfun$blockForResult$2(TaskRunSyncUnsafe.scala:150)
  at monix.eval.internal.TaskRunSyncUnsafe$$$Lambda$400/0x0000000800ea2840.apply$mcZ$sp(Unknown Source)
  at scala.runtime.java8.JFunction0$mcZ$sp.apply(JFunction0$mcZ$sp.java:23)
  at monix.execution.internal.Trampoline$$anon$1.blockOn(Trampoline.scala:84)
  at scala.concurrent.package$.blocking(package.scala:146)
  at monix.eval.internal.TaskRunSyncUnsafe$.blockForResult(TaskRunSyncUnsafe.scala:150)
  at monix.eval.internal.TaskRunSyncUnsafe$.apply(TaskRunSyncUnsafe.scala:97)
  at monix.eval.Task.runSyncUnsafeOpt(Task.scala:1096)
  at monix.eval.Task.runSyncUnsafe(Task.scala:1063)
  at enqueue(xxx.scala:39)

в 0.4.0 моникс 3.0.0, в 0.7.1 - 3.1.0

Блядь, как же уебищно отформатировалось

Моникс знач

🤔 давно на таком сетапе и пока не залочилось

именно обсерваблы обсеренькались мб

Ну очень странно, на 3.0.0 всё заебок, вот на 3.1.0 похоже не заебок. Причем сидит оно именно на sync.onNext курит. Казалось бы - це ж буффер, кладешь в него - есть место - положилось, нет места - дропает новый элемент. С другого конца эта очередь разбирается отдельным потоком на отдельном пуле. Да, скорее всего, по разным причинам 90% времени этот буфер будет заполнен, ну так и пусть он новые дропает, чего ждать-то

а какая стратегия?

а вижу

Task.delay(sync.onNext(entry)).startAndForget.runSyncUnsafe(1.minute)
сложна

почему не просто sync.onNext(entry) ?

Это часть хуйни, которая отсылает логи куда-то там, она описана как наследник логбековской хуйни. И апи там синхронное, соответственно

Хотя это не отвечает на твой вопрос, да

Так

тебе возвратили синхронный обсиратор

казалось бы обсирай синхронно

зачем оборачивать в таск и потом лочиться

Кажется, что Observer.Sync.onNext и так гарантирует синхронно вернуть Ack