Signaling from the sink implementation is an unnecessary complication that it (the sink) shouldn't have to deal with. For now, I managed to serialize running of the sink and running ofrepeatTask insiderepeatAndFeedStatus with the help of anMVar and aSemaphore; it's not elegant (and possibly not performant), but it gets the job done in an encapsulated way (this implementation detail is nicely hidden insiderepeatAndFeedStatus).

I already considered using a ZIO instead of a sink (it's in the original question), but as I stated, if I do this I lose the semantics of repetition and sequentiality that streams/sinks have. So it's not an optimal solution in terms of expressiveness.

So, back to the original question: is it not possible to feed items into aZSink manually?

You can feed items to a sink manually by offering them to a queue.ZStream.fromQueue(queue).run(sink).fork and thenqueue.offer.

You can feed items to a sink manually by offering them to a queue.ZStream.fromQueue(queue).run(sink).fork and thenqueue.offer.

Well, that's what I'm already doing, but it doesn't look manual to me because it's the stream that does the actual feeding, in its own fiber. (I suppose I should have written "directly" instead of "manually".) And it doesn't have the properties that I was hoping for. :(

Anyway, thanks for your input.

Yeah, I think you are looking for this further property that writing to the sink doesn't return until the sink has "completed" processing that element and the most that could be guaranteed is that the sink has "started" processing the element. I feel like this problem could be more easily solved withZIO or possibly withSchedule.

Yeah, I think you are looking for this further property that writing to the sink doesn't return until the sink has "completed" processing that element and the most that could be guaranteed is that the sink has "started" processing the element.

Yes, this may be the case. I did further testing of the solution that I posted in my answer, and when I add a delay to the sink implementation, it doesnot currently work like I want it to. Maybe your other answer will lead to a better result.

So it's using aStream all the way... I wasn't sure this was possible (and I'm still not sure that I can make a version that also does error handling/retrying in addition to normal repetition) and I'm not familiar with the way streams are composed, but it may be worth a try.

I managed to implement the full functionality of the function via streams, as suggested:

defrepeatAndFeedStatus2(initTask :Task[Unit],repeatTask :Task[Unit],repeatDelay :Duration,retrySchedule :Schedule[Any,Throwable,Unit],statusSink :Sink[Throwable,RunStatus,Nothing,Nothing]):Task[Nothing]={valrepeatSchedule:Schedule[Any,Any,?]=Schedule.fixed(repeatDelay)Ref.make(false).flatMap { (failedRef :Ref[Boolean])=>valloopStream:ZStream[Any,Throwable,RunStatus]=ZStream.whenZIO(failedRef.get) {ZStream.succeed(RunStatus.Recovering) }++ZStream.fromZIO(initTask).drain++ZStream.repeatZIOWithSchedule(repeatTask*>ZIO.succeed(RunStatus.Running), repeatSchedule)++ZStream.fail(IllegalStateException("Repeat isn't expected to complete"))valloopWithRetryStream:ZStream[Any,Throwable,RunStatus]=      loopStream        .either        .flatMap[Any,Throwable,RunStatus] {caseRight(runStatus)=>ZStream.succeed(runStatus)caseLeft(ex)=>ZStream.fromZIO(failedRef.set(true)).drain++ZStream.succeed(RunStatus.Failing(ex,Instant.now()))++ZStream.fail(ex)        }        .retry(retrySchedule)    (ZStream.succeed(RunStatus.Starting)++ loopWithRetryStream).run(statusSink)  }}

This seems to be a satisfactory solution to the specific problem that I faced. It doesn't use additional state or parallelism, which is definitely a win. Error handling is unwieldy, but not too much.