original.kt

class Test {
    @Volatile
    var ticker = 0

    @Volatile
    var deferred: Deferred<String>
    val retryingExceptionHandler: CoroutineExceptionHandler = CoroutineExceptionHandler { _, throwable ->
        println("!!!!!!!")
        authScope.kickoffUpdateWithDelay()
        authScope.testB()
    }
    val authScope = CoroutineScope(SupervisorJob() + Dispatchers.IO + retryingExceptionHandler)

    init {
        deferred = authScope.async { getSome() }
        authScope.testB()
    }

    fun CoroutineScope.kickoffUpdateWithDelay() = launch {
        delay(100)
        deferred = authScope.async { getSome() }
    }

    suspend fun getSome(): String {
        if (ticker < 5) {
            delay(1)
            ticker += 1
            throw Exception()
        } else {
            delay(1)
            ticker += 1
            return "PASSSED"
        }
    }

    fun CoroutineScope.testB() = launch {
        deferred.await()
        if (ticker < 5) {
            ticker += 1
            throw Exception()
        } else {
            ticker += 1
            println("PASSSED")
        }
    }
}

fun main(args: Array<String>) = runBlocking<Unit> {

    Test()

    repeat(100) {
        delay(100)
    }
}

updated.kt

@ExperimentalCoroutinesApi
class _Test {

    val mutex = Mutex()
    data class State(val ticker: Int, val producerJob: Job?){
        fun increment() = copy(ticker = ticker + 1)
    }

    val results = ConflatedBroadcastChannel<String>()

    @Volatile
    var state = State(0, null)

    val retryingExceptionHandler: CoroutineExceptionHandler = CoroutineExceptionHandler { _, throwable ->
        println("!!!!!!!")
        authScope.kickoffUpdateWithDelay()

        authScope.testB()
    }
    val authScope = CoroutineScope(SupervisorJob() + Dispatchers.IO + retryingExceptionHandler)

    init {
        require(mutex.tryLock())
        try {
            state = state.copy(producerJob = authScope.launch { getSome() })
        }
        finally {
            mutex.unlock()
        }
    }

    fun CoroutineScope.kickoffUpdateWithDelay() = launch {
        delay(100)
        mutex.withLock {
            if(state.producerJob?.isCancelled == true){
                state = state.copy(producerJob = authScope.launch { getSome() })
            }
        }
    }

    suspend fun getSome(): Unit {
        if (state.ticker < 5) {
            delay(1)
            mutex.withLock { state = state.increment() }
            throw Exception()
        } else {
            delay(1)
            mutex.withLock { state = state.increment() }

            results.send("PASSSED")
        }
    }

    fun CoroutineScope.testB() = launch {

        state.producerJob?.join()
        //important difference between join() and await():
        // await() will throw the exception thrown by the block of the deferred,
        // where join() will not.

        if (state.ticker < 5) {
            mutex.withLock { state = state.increment() }
            throw Exception()
        } else {
            mutex.withLock { state = state.increment() }
            results.send("testB: PASSSED")
        }
    }
}

fun main(args: Array<String>) = runBlocking<Unit> {

    _Test()

    repeat(100) {
        delay(100)
    }
}