Не совсем то, на самом деле
Я рассматривал вариант с использованием withTimeout
Основная проблема именно в прерывании этого таймаута насильно, грубо говоря

То есть как это например реализовано в телеграме (визуально):
Мы не можем подключиться, пытаемся через 10 сек
Мы не можем подключиться, пытаемся через 30 сек
Мы не можем подключиться, пытаемся через 60 сек
Попробовать подключиться (нажимаем)
О, подключились

То есть грубо говоря, таймаут там все равно стоит, на попытку подключения, но его можно прерывать насильно
Не очень понимаю как это можно сделать с withTimeout или delay

Сейчас это сделано очень костыльно - с использованием async, в котором находится lock.wait(time), и я извне могу сделать notifyAll
Но это слишком костыльно

Примерно, как это выглядит:

Само ожидание

    private fun CoroutineScope.asyncWait(timeout: Long, units: TimeUnit) = async {
        synchronized(lock) {
            try {
                lock.wait(units.toMillis(timeout))
            } catch (ex: TimeoutException) {
            }
        }
    }

GlobalScope.launch {
   while(true) {
      if(tryCounts.get() >= 3) {
          Logger.info("Waiting 45 seconds")
          asyncWait(45, TimeUnit.SECONDS).await()
       }

       try {
           ... отправка

          tryCounts.set(0)
       } catch(ex: Exception) {
          ex.printStackTrace()
          tryCounts.incrementAndGet()
       }
   }
}

Но, как и было сказано ранее, это прям суперкостыльное решение, за которое по хорошему себе бы руки отбить

synchronized и корутины — несовместимые штуки.
Про прервать насильно так и не понял. Если запрос отработает раньше таймаута, всё будет хорошо. Если таймаут наступит раньше, корутина будет отменена. А, ну да, если запрос блокирующий, то сам запрос будет висеть. Но если нет апишки на отмену блокирующего запроса, то всё будет плохо и так.

Что вы хотите-то? Вообще по внешнему запросу отменить ожидание?

    private fun CoroutineScope.asyncWait(timeout: Long, units: TimeUnit) = async {
        synchronized(lock) {
            try {
                lock.wait(units.toMillis(timeout))
            } catch (ex: TimeoutException) {
            }
        }
    }

GlobalScope.launch {
   while(true) {
      if(tryCounts.get() >= 3) {
          Logger.info("Waiting 45 seconds")
          asyncWait(45, TimeUnit.SECONDS).await()
       }

       try {
           ... отправка

          tryCounts.set(0)
       } catch(ex: Exception) {
          ex.printStackTrace()
          tryCounts.incrementAndGet()
       }
   }
}

Можно разделить само переподключение и ожидание перед автоматическим переподключением. Вторую задачу можно легко отменять при необходимости (ручном запросе).

Отменить ожидание по внешнему запросу. Да. Сейчас, в текущем коде, отмена происходит следующим образом:

fun wakeUp() { 
        synchronized(lock) {
            lock.notifyAll()
        }
}

asyncWait сразу же прерывается и совершается следующая попытка отправки

Если вопрос во внешней отмене, то можно использовать select, но я так и не научился им пока пользоваться.

Каким образом можно отменить ожидание? Можно поподробнее?
По сути, если я правильно понимаю, у меня сейчас как раз и разделено ожидание и попытка переподключения
Но только полностью неправильно

Либо как вариант просто передать джобу и проверять её руками, а снаружи кэнцеллить.

Если это отдельная Job - через cancel().
Но там тоже есть ньюансы (нужен supervisor).
Честно говоря, я не знаю, как это с ходу сделать красиво без костылей, тут думать надо :)

А с виду очень даже простая задача... Пока не лезешь глубже :D

Что то не могу понять это баг или фича:

fun main() {
    
    class Foo {
        operator fun invoke(f: (Int) -> Int): Int = f(1)
    }
    
    fun foo() = Foo()
    foo() { it + 1 }
}

Так не компилируется

А так компилируется

foo()() { it + 1}
foo().invoke { it + 1}

Кажется, все логично

foo не принимает же лямбду первым аргументом

при этом, так тоже компилируется:

val foo = Foo()
foo { it + 1}

Так то да, ну вот я покопал доки и не понял это баг или фича

Фича. Обычная работа invoke'a и лямбд
invoke, когда не является статическим, просто позволяет обращаться к объекту как к функции
foo { ... } - вызов функции invoke с аргументом лямбдой

Он получается всегда пытается пихнуть лямбду в предыдущую скобку, shit

Не совсем