Made block streaming interruptible

[vsubhuman]

Added internal functionality that allows to interrupt block-streaming when necessary. But it's not used for now. Required for later rollback processing.

Part of https://github.com/input-output-hk/rust-cardano/pull/693#issuecomment-494574901

[vsubhuman]

@vincenthz @NicolasDP

[mzabaluev]

In grpc, we have an API producing futures-0.1 streams for incoming blocks. Interrupting there is the matter of dropping the stream. But this is the old protocol crate that does not do futures?

[vsubhuman]

But this is the old protocol crate that does not do futures?

@mzabaluev, yes, it seems that way, cuz the protocol used from the Bridge only accepts a callback function that gets called with every new block and only previously returned Result<()>, so the only existing option for any interaction between the protocol and a protocol caller is for caller to return an Error. Which might be an alternative to this whole proposed scheme

[vsubhuman]

so the only existing option for any interaction between the protocol and a protocol caller is for caller to return an Error. Which might be an alternative to this whole proposed scheme

But also requires changes, because right at the moment any result of the callback lambda is just ignored somewhere in between a caller and the protocol: https://github.com/input-output-hk/rust-cardano/pull/705/files#diff-009abb88e20c2ce711e56e1a1bbd482dL220

[vsubhuman]

@vincenthz @NicolasDP @mzabaluev, is it ok, or should we switch to using Result<()> and returning a special kind of error to interrupt?

[mzabaluev]

Result is more conventional. It also corresponds to the streamed protocol implementation, for which we signal premature stream end with some kind of error.

[mzabaluev]

@vantuz-subhuman Did you decide that you don't want it?

[vsubhuman]

@mzabaluev, sorry for a delayed response (vacation 🙂). We are currently working on reworking the interruption logic in our code, to use Result<()> and errors instead. Once we are done - I will open another PR

[vsubhuman]

@mzabaluev, so far we have only one major problem with switching to use errors - this function call: https://github.com/input-output-hk/rust-cardano/blob/68d8d586f0247b89c91b9cfc98b828d8f59274ad/exe-common/src/network/native.rs#L214-L223

native impl from exe-commons calls stream_blocks function from "deeper" protocol package. The stream_blocks function takes a lambda that can return protocol::Error, because it's a function from protocol package, but when we pass that lambda here we call the got_blocks closure - https://github.com/input-output-hk/rust-cardano/blob/68d8d586f0247b89c91b9cfc98b828d8f59274ad/exe-common/src/network/native.rs#L220 which we are now changing to be able to return network::Error.

The problem is that network::Error is not universally convertible into protocol::Error (rather otherwise, thru wrapper network::Error::ProtocolError(_)). So if internal lambda that we implement in sync.rs returns network:Error - there's no nice way to pass it thru the stream_blocks call.

One possible solution is to make higher level lambda from Api::get_blocks to only return protocol::Error, which logically makes sense to me. Do you think it will it be an ok solution?

[mzabaluev]

A brief look suggests that a closure that is passed to stream_blocks might return Result<_, protocol::Error>, and that would be propagated to the return value of stream_blocks. If that's not good, you could invent a special purpose error type just for this callback and convert it to an error of stream_blocks as appropriate.

get_blocks is the higher level method and it's appropriate that it returns network::Error to be able to report all other classes of failures that may occur.

[vsubhuman]

get_blocks is the higher level method and it's appropriate that it returns network::Error to be able to report all other classes of failures that may occur.

@mzabaluev, that's the problem, that stream_blocks can only return protocol::Error, which does not know and does not cover all the error cases that might be implied by network::Error =(

A diagram of scope here looks like this:

function that can return network::Error {

  result = function that can return protocol::Error {

    internal_result = function that can return network::Error {
      ...
    }

    if internal_result is network::Error {
      // how to return protocol::Error from network::Error ???
    }

    return Ok()

  }

  if result is protocol::Error {
    return network::Error::ProtocolError(result)
  }

  return result
}

Logically network::Error is > than protocol::Error, and network::Error can BE protocol::Error thru use of network::Error::ProtocolError(...) wrapper:

So if internal function call returns higher level network::Error of larger scope - I don't see any nice way on how we can "downgrade" it to lover level protocol::Error of smaller scope, without making weird cyclical dependencies and creating something like a protocol::Error::NetworkError(...) wrapper, which then would allow abomination like network::Error::ProtocolError( protocol::Error::NetworkError( .... ) ), etc.

Calling lover level protocol function with a callback that can return higher level network error creates this weird situation where we gotta try to squeeze a wider type thru a narrower type ☹️ Like trying to pass integer numbers thru a function that can only return natural numbers )

The only possible solutions I see is to either narrow potential error type for get_blocks so it can fit thru the narrower type of the underlying protocol calls, or to widen the underlying protocol error types, so, for example, stream_blocks could potentially return any type as error. Do you mean this by "invent a special purpose error type"? 🤔 Because any error type supported by lower level stream_blocks has to be implemented on the same low level, so it cannot directly know about higher level network::Error. Maybe there could be a special protocol::Error::StreamingCallbackError(Any) so stream_blocks still could return protocol errors but would have a special kind of wrapper to return any wider callback error? The callback error handling on the higher level will look kinda awkward then, tho

[vsubhuman]

The callback error handling on the higher level will look kinda awkward then, tho

Kinda like:

result = function that can return protocol::Error {
}

if result is protocol::Error {
  if result is protocol::Error::StreamingCallbackError( some ) {
    if some is network::Error {
      return some as network::Error
    }
    return network::Error::SomeError( some )
  }
  return network::Error::ProocolError( result )
}
return result

[vsubhuman]

Underlying stream_blocks doesn't actually do anything with the callback error and just returns it immediately, so other option could be to make stream_blocks callback implicit in the error type, but then stream_blocks itself has to return a type that imply containing either protocol::Error or generic type error.

Something like:

fn stream_blocks<F, E> ( got_blocks: &mut F ) -> Result<(), Either<protocol::Error, E>>
where F: FnMut(RawBlock) -> Result<(), E>

This prolly can be done with a custom type, like:

enum StreamBlocksError<E> {
  ProtocolError(protocol::Error),
  CallbackError(E)
}

@mzabaluev, is this what you meant by a special purpose error type? 🤔

fn stream_blocks<F, E> ( got_blocks: &mut F ) -> Result<(), StreamBlocksError<E>>
where F: FnMut(RawBlock) -> Result<(), E>

[vsubhuman]

Other option could be to just reopen this PR that uses returning an enum with two options, tho 😁

[mzabaluev]

@vantuz-subhuman Is it not practical to extend protocol::Error with the variant specifically for the streaming callback failure? I also don't get why the got_blocks callback must return fully rich network::Error (appreciate the Venn diagram), rather than just something really small and topical. In the rescinded change it was just the unit-like BlockReceivingFlag::Stop value that did the job, right?

[vsubhuman]

I also don't get why the got_blocks callback must return fully rich network::Error (appreciate the Venn diagram), rather than just something really small and topical.

Yeah, that makes sense to me, that's why I initially proposed that got_blocks callback return type should be changed from returning network::Error to return protocol::Error, so underlying protocol call can pass it easily. And then protocol::Error can be extended with any logical kind of error, e.g. protocol::Error::Rollback.

If that's ok to you - I will do it like this.

[mzabaluev]

So it could be e.g.

#[derive(Debug)]
struct StopHammerTime;

fn stream_blocks<F> ( got_blocks: &mut F ) -> Result<(), protocol::Error>
where F: FnMut(RawBlock) -> Result<(), StopHammerTime>

[vsubhuman]

It seems that I have confused your response:

get_blocks is the higher level method and it's appropriate that it returns network::Error to be able to report all other classes of failures that may occur.

To mean that got_blocks callback should return network::Error. If that's not the case - then we are on the same page )

[mzabaluev]

Oh BTW, the closure does not need to be passed under a mutable reference. The mutable data capture is already baked into the generic parameter. This would do:

fn stream_blocks<F> (got_blocks: F) -> Result<(), protocol::Error>
where F: FnMut(RawBlock) -> Result<(), StopHammerTime>

(also remember to cargo fmt before sending your changes)

[vsubhuman]

Yeah, makes sense to me. Also realised that making callback to return protocol::Error does not make much sense, since it's only one of the implementations, and callback return type is abstract in Api module =\

My bad.

I'm going with something like this, declared in the Api module:

Thank you for the support! 🙌