[BUG] Cannot use a body in serverSecurityLogic: causes java.lang.IllegalStateException: This publisher only supports one subscriber

[samthebest]

Tapir version: 1.11.9

Scala version: 2.13.15

Suppose we have some security logic, nice and neatly decoupled and deduplicated from the business logic:

val secureBase =
     ...
        .securityIn(extractFromRequest(_.uri))
        .securityIn(extractFromRequest(_.method))
        .securityIn(stringBody)
        .serverSecurityLogic { case (jwt, basic, uri, method, body) => ... Authed(...) }

Then somewhere we try to build on this to add the business logic:

secureBase
  .in(jsonBody[Person])
  .serverLogic(authed => person => ...)

Then hitting this endpoint results in:

java.lang.IllegalStateException: This publisher only supports one subscriber
	at org.playframework.netty.HandlerPublisher.subscribe(HandlerPublisher.java:167)
	at org.playframework.netty.http.DelegateStreamedHttpRequest.subscribe(DelegateStreamedHttpRequest.java:19)
	at sttp.tapir.server.netty.cats.internal.Fs2StreamCompatible$$anon$1.$anonfun$fromPublisher$2(Fs2StreamCompatible.scala:52)
	at flatMap @ fs2.Compiler$Target.flatMap(Compiler.scala:163)
	at flatMap @ fs2.Compiler$Target.flatMap(Compiler.scala:163)
	at flatMap @ fs2.Pull$.fs2$Pull$$interruptGuard$1(Pull.scala:951)
	at flatMap @ fs2.Compiler$Target.flatMap(Compiler.scala:163)
	at flatMap @ fs2.Compiler$Target.flatMap(Compiler.scala:163)
	at flatMap @ fs2.Pull$.fs2$Pull$$interruptGuard$1(Pull.scala:951)
	at flatMap @ fs2.Compiler$Target.flatMap(Compiler.scala:163)
	at flatMap @ fs2.Compiler$Target.flatMap(Compiler.scala:163)
	at flatMap @ fs2.Pull$.fs2$Pull$$interruptGuard$1(Pull.scala:951)
	at async_ @ fs2.interop.reactivestreams.StreamSubscriber$$anon$1.dequeue1(StreamSubscriber.scala:219)

I think what is needed in Tapir to resolve this is to cache the body, so that it can be read for security logic, and also for business logic.

Of course one could argue that an API that uses the body for auth isn't truly RESTful? But still, doesn't seem too naughty.

In the meantime I was hoping there is some neat work around? The only work arounds I've come up with means bypassing the .in(jsonBody part of the DSL 😬 , i.e. by storing the body in Authed then parsing it out later with some sugar:

  implicit class AuthzdEndpointSyntax[INPUT, OUTPUT](endpoint: PSE[INPUT, OUTPUT]) {
    def serverLogicBodyIn[T: Decoder](f: (Authed, INPUT, T) => IO[Either[ErrorInfo, OUTPUT]]) = ...

I was hoping there is a flag somewhere I can set justReadBodyTwicePlease

[adamw]

Interesting case :) I think for now the work-around you describe it the way to go.

But we probably could support this better, although there are some challenges as well:

1. as a minimum, we should detect that there are double body inputs in the endpoint, and throw a proper exception, not a weird publisher one
2. if we supported multiple-body inputs, what about client interpreters? there, the inputs are given as values - which body should be chosen to be sent over the network? Although, truth be told, I'm not sure what would happen if we used the same query parameter twice ;). Maybe some way around this is to provide something similar to extractFromRequest (which is ignored during client generation) to extract the request body again
3. again if we supported multi-body inputs, we would have to verify that they all have a "replayble" type. E.g. two inputs which are byte arrays are fine, but two streaming bodies, or a streaming body combined with a byte array one, is impossible (as you can either read everything into memory, or stream lazily)

[samthebest]

Ah, I knew removing .in(jsonBody was going to have some bad consequences:

• We lose auto generated docs being correct
• We lose restish working nicely

Before removing .in(jsonBody

After removing .in(jsonBody

We have a bold team member that might have a crack at forking NettyCatsServer to work in a simpler way - i.e. just treat the body as an immutable byte array in memory by decoupling from fs2. So it's not multiple-body inputs, it's one-body used in the serverSecurityLogic and the same body used in the serverLogic. Though I'd be surprised @adamw if you hadn't seen that done before? Save him a fun weekend? 😂

two inputs which are byte arrays are fine, but two streaming bodies ...

It does feel like catering to APIs that want streaming is an inversion of desired generality. Most APIs just have json bodies that don't need streaming. I think the industry has gone a bit wild and wacky and forgotten that HTTP stands for Hyper-Text transfer protocol. If people want to stream data between two backend web services, they shouldn't be using HTTP, they really should be using message queues or topics.

That said, we probably shouldn't be using body's for auth logic 😅 . We're not locked into a big ugly enterprise RBAC model fortunately.

Anyway, we're going to ditch this work around: def serverLogicBodyIn[T: Decoder](, and just change our API to stop using the body as part of security logic by splitting up endpoints into more endpoints. Some may argue that's more RESTful? 💪

[adamw]

Re: streaming - regardless of our opinion on streams, if we do go with the route of supporting multiple body definitions, an endpoint which requests the body both as a stream and a string is simply invalid - and we can only report it during run-time (and we should, which we don't do right now).

As for the implementation itself. Apart from clients, documentation is also an issue - just as you write. If you have a stringBody and jsonBody[T], which one do you use for docs? Or if you have a stringBody and a byteArrayBody?

So I think the right route to implement this would be with something like extractFromRequest(_.body), which would be the "secondary" body definition, not included in the docs, discarded for clients.

Changes in NettySyncServer are only part of the story, though, as you'll also need to change ServerInterpreter and/or ServerRequest. But I do welcome even draft PRs :)

[samthebest]

Yes, thanks @adamw

So I think the right route to implement this would be with something like extractFromRequest(_.body), which would be the "secondary" body definition, not included in the docs, discarded for clients.

Absalutely! Perhaps even deprecate stringBody entirely as something that can be used in serverSecurityIn