zio/interop-reactive-streams

Fiber failed. A checked error was not handled.

asakaev opened this issue · 8 comments

asakaev commented

Getting failed fiber when consuming a Stream (converted from Publisher). Works fine by the way.

Fiber failed.
A checked error was not handled.
com.mongodb.MongoWriteException: E11000 duplicate key error collection: db.collection index: _id_ dup key: { _id: "42" }

Fiber:Id(1577203251494,3) was supposed to continue to:
  a future continuation at zio.ZIO.ensuring(ZIO.scala:350)
  a future continuation at zio.ZIO.ensuring(ZIO.scala:350)

Fiber:Id(1577203251494,3) execution trace:
  at zio.ZIO$._IdentityFn(ZIO.scala:2630)
  at zio.Promise.await(Promise.scala:49)
  at zio.ZIOFunctions.effectAsyncInterrupt(ZIO.scala:1952)
  at zio.ZIOFunctions.effectAsyncInterrupt(ZIO.scala:1952)

Fiber:Id(1577203251494,3) was spawned by:

Fiber:Id(1577203251199,1) was supposed to continue to:
  a future continuation at zio.ZManaged.flatMap(ZManaged.scala:306)
  a future continuation at zio.ZManaged.flatMap(ZManaged.scala:307)
  a future continuation at zio.stream.ZStream.run(ZStream.scala:2051)
  a future continuation at zio.ZIO.run(ZIO.scala:1166)
  a future continuation at zio.ZManaged.use(ZManaged.scala:748)
  a future continuation at shared.Scratch$.application(Scratch.scala:36)
  a future continuation at shared.Scratch$.run(Scratch.scala:44)
  a future continuation at shared.Scratch$.run(Scratch.scala:45)

Fiber:Id(1577203251199,1) execution trace:
  at zio.ZManaged.flatMap(ZManaged.scala:305)
  at zio.Ref$.update$extension(Ref.scala:123)
  at zio.ZManaged.flatMap(ZManaged.scala:305)
  at zio.ZManaged.flatMap(ZManaged.scala:307)
  at zio.Ref$.update$extension(Ref.scala:123)
  at zio.ZManaged.flatMap(ZManaged.scala:307)
  at zio.ZManaged.flatMap(ZManaged.scala:301)
  at zio.Ref$.make(Ref.scala:169)
  at zio.ZManaged.flatMap(ZManaged.scala:306)
  at zio.internal.FiberContext$InterruptExit$.apply(FiberContext.scala:150)

Fiber:Id(1577203251199,1) was spawned by:

Fiber:Id(1577203251108,0) was supposed to continue to:
  a future continuation at zio.App.main(App.scala:57)
  a future continuation at zio.App.main(App.scala:56)

Fiber:Id(1577203251108,0) ZIO Execution trace: <empty trace>

Fiber:Id(1577203251108,0) was spawned by: <empty trace>
runtologist commented

Could you add a bit of information about the code that leads to the error? Maybe there is a way of providing a minimal example? The trace do not mention interop-reactive-streams, so it is a bit difficult to work on this. :) From the trace it looks like a failure in Mongo simply bubbled up to your business logic, which probably it the right thing, but I can't tell without more context.

asakaev commented 
db.getCollection(collection)
  .replaceOne(
    Filters.and(Filters.eq(Field.Id, id), Filters.eq(Field.Version, version)),
    new Document()
      .append(Field.Id, id)
      .append(Field.Version, version + 1)
      .append(Field.Payload, payload),
    new ReplaceOptions().upsert(true)
  )
  .toStream(1)
  .runHead

This leads to:

com.mongodb.MongoWriteException: E11000 duplicate key error collection: db.numbers index: _id_ dup key: { _id: BinData(0, 00000001) }

I am trying to implement optimistic concurrent update with version. If Id and Version are correct, then update payload and bump version.

If such document is not found then try to create it. I can catch MongoWriteException in .runHead.foldM(...) and handle it, but I believe there should be no failed Fiber.

Maybe there is some place where this exception could me managed before it going to stderr?Correct me if I am wrong.

runtologist commented

Unless the error get's handled somewhere the Fiber will fail. The only alternative would be swallowing the error I think. Or do you see a failed Fiber even though the error gets handled? In that case, could you provide an example including error handling?

runtologist commented

For building optimistic concurrency control - how about using ZStream.orElse to fallback to create mode?

runtologist commented

@asakaev is there any way I can help you get forward with this?

asakaev commented

Simon, it looks like I discovered a minimal sample to produce behavior that seems mysterious for me. Not sure that it's reactive streams interop, more like zio runtime. I can not figure out where this exception occurs and why it falls in logs while still correctly handled by zio.

https://github.com/asakaev/mongo-pipe/blob/exception/src/main/scala/com/github/asakaev/ExceptionExampleApp.scala

Briefly: I create a MongoDB client and try to stream all documents from the collection. I expect that the stream will fail with MongoTimeoutException if I do this with an unreachable server. Behavior is correct, but I see the wall of Fiber failed logs.

SLF4J: Failed to load class "org.slf4j.impl.StaticLoggerBinder".
SLF4J: Defaulting to no-operation (NOP) logger implementation
SLF4J: See http://www.slf4j.org/codes.html#StaticLoggerBinder for further details.
Fiber failed.
A checked error was not handled.
com.mongodb.MongoTimeoutException: Timed out after 1 ms while waiting for a server that matches ReadPreferenceServerSelector{readPreference=primary}. Client view of cluster state is {type=UNKNOWN, servers=[{address=127.0.0.1:27017, type=UNKNOWN, state=CONNECTING, exception={com.mongodb.MongoSocketOpenException: Exception opening socket}, caused by {java.net.ConnectException: Connection refused}}]
  at com.mongodb.internal.connection.BaseCluster.createTimeoutException(BaseCluster.java:408)
  at com.mongodb.internal.connection.BaseCluster.handleServerSelectionRequest(BaseCluster.java:316)
  at com.mongodb.internal.connection.BaseCluster.access$800(BaseCluster.java:66)
  at com.mongodb.internal.connection.BaseCluster$WaitQueueHandler.run(BaseCluster.java:491)
  at java.lang.Thread.run(Thread.java:748)

Fiber:Id(1580652095344,3) was supposed to continue to:
  a future continuation at zio.ZIO.ensuring(ZIO.scala:350)
  a future continuation at zio.ZIO.ensuring(ZIO.scala:350)

Fiber:Id(1580652095344,3) execution trace:
  at zio.ZIO$._IdentityFn(ZIO.scala:2630)
  at zio.Promise.await(Promise.scala:49)
  at zio.ZIOFunctions.effectAsyncInterrupt(ZIO.scala:1952)
  at zio.ZIOFunctions.effectAsyncInterrupt(ZIO.scala:1952)

Fiber:Id(1580652095344,3) was spawned by:

Fiber:Id(1580652093859,1) was supposed to continue to:
  a future continuation at zio.ZManaged.flatMap(ZManaged.scala:306)
  a future continuation at zio.ZManaged.flatMap(ZManaged.scala:307)
  a future continuation at zio.stream.ZStream.catchAllCause(ZStream.scala:845)
  a future continuation at zio.stream.ZStream.catchAllCause(ZStream.scala:846)
  a future continuation at zio.stream.ZStream.catchAllCause(ZStream.scala:850)
  a future continuation at zio.stream.ZStream.run(ZStream.scala:2045)
  a future continuation at zio.ZIO.run(ZIO.scala:1166)
  a future continuation at zio.ZManaged.use(ZManaged.scala:748)
  a future continuation at com.github.asakaev.ExceptionExampleApp$.run(ExceptionExampleApp.scala:38)

Fiber:Id(1580652093859,1) execution trace:
  at zio.ZManaged.flatMap(ZManaged.scala:305)
  at zio.Ref$.update$extension(Ref.scala:123)
  at zio.ZManaged.flatMap(ZManaged.scala:305)
  at zio.ZManaged.flatMap(ZManaged.scala:307)
  at zio.Ref$.update$extension(Ref.scala:123)
  at zio.ZManaged.flatMap(ZManaged.scala:307)
  at zio.ZManaged.flatMap(ZManaged.scala:301)
  at zio.Ref$.make(Ref.scala:169)
  at zio.ZManaged.flatMap(ZManaged.scala:306)
  at zio.internal.FiberContext$InterruptExit$.apply(FiberContext.scala:150)

Fiber:Id(1580652093859,1) was spawned by:

Fiber:Id(1580652093592,0) was supposed to continue to:
  a future continuation at zio.App.main(App.scala:57)
  a future continuation at zio.App.main(App.scala:56)

Fiber:Id(1580652093592,0) ZIO Execution trace: <empty trace>

Fiber:Id(1580652093592,0) was spawned by: <empty trace>
Stream.failed: Traced(Fail(com.mongodb.MongoTimeoutException: Timed out after 1 ms while waiting for a server ...

Process finished with exit code 0
runtologist commented

Thanks for the example, that clarifies a lot. So everything works as expected. The error also gets propagated. However, an additional Fiber failed gets logged.

I guess I found the reason: A Fiber gets started, which watches for completion (success or failure) of the reactive stream and shuts down the queue used to feed the ZStream. It uses Promise.await for this, so it also fails when the Promise fails. The fiber still behaves correctly (using ensuring), but the failure gets logged.

I'll try to reproduce that in a unit test and provide a fix later.

  • 👍1
runtologist commented

Fixed in v1.0.3.5-RC3. Thanks @asakaev for reporting this bug and helping to track it down.

  • ❤️2