Implement Ingress Service

[slinkydeveloper]

Fix #452

This PR is a first iteration at implementing the Ingress service, to execute background calls from the ingress.

[slinkydeveloper]

Admittedly, this is not the finest of the implementations, but it works fine. I'll take a second pass at this code with the JsonMapper introduced by #43, which should avoid leaking the "special handling" of InvokeRequest from the protocol mod.

Will PR an e2e test soon.

[slinkydeveloper]

Tested locally with restatedev/e2e#150 and it works

[slinkydeveloper]

One problem with the current implementation approach is that there's no "straightforward" way to allow the call to be a delayed call, because there's no way to set an arbitrary timer from the ingress to the partition processor, given that first of all you need to pick a partition processor.

I wonder whether we should follow another implementation approach where dev.restate.Ingress is implemented as built-in service, thus letting the partition processor handle it.

Pros:

• Easily implement the delayed call, possibly other features in future
• The user can invoke dev.restate.Ingress from within another function. This is not particularly useful, but right now it's kinda weird to have this distinction of services that can be accessed only from the ingress.
• Removes the hacks around the method descriptor registry

Cons:

• One extra hop to start the invocation

WDYT @tillrohrmann?

[tillrohrmann]

Thanks for creating this PR @slinkydeveloper. The changes look good to me. +1 for applying the improvements once we have the schema registry in place. I think you are also right that for supporting delayed calls we probably need to move the processing of the Ingress/Invoke calls to a partition processor.

While trying things out, I noticed that the dev.restate.Ingress/Invoke service does not get automatically registered. That's why I couldn't initially invoke it. After hacking an automatic registration at the MethodDescriptorRegistry, I could invoke services via connect and protobuf.

[tillrohrmann]

How will the dev.restate.Ingress service be registered at the MethodDescriptorRegistry?

[slinkydeveloper]

I'm sorry, i forgot to push the last version of the branch :( check now

[tillrohrmann]

Where I stumbled upon a bit was where are we checking whether a protobuf invoked dev.restate.Ingress/Invoke call actually calls an existing service. When using connect, this happens in the read_json_invoke_request method quite explicitly. For protobuf encoded dev.restate.Ingress/Invoke requests it was not that obvious.

It turns out that we are doing this check in the invocation_factory.create call where we try to extract the key. If the service is not registered, then it fails with an UnknownService error.

Due to the different checks, we do see different manifestations of the unknown service:

2023-06-22T10:45:16.459608Z WARN restate_ingress_grpc::handler
  Cannot create service invocation: UnknownServiceMethod { service_name: "counter.Counter", method_name: "Foobar" }
  in restate_ingress_grpc::handler::ingress_service_invocation
    rpc.system: "grpc"
    rpc.service: dev.restate.Ingress
    rpc.method: Invoke
2023-06-22T10:45:37.919692Z WARN restate_ingress_grpc::protocol::connect_adapter
  Error when parsing request dev.restate.Ingress.Invoke: status: NotFound, message: "counter.Counter/Foobar not found", details: [], metadata: MetadataMap { headers: {} }

I think it is not a problem just that I tripped over the fact that I didn't find where the check for an existing method in the protobuf case is done.

[tillrohrmann]

Thanks for updating this PR. LGTM with the InMemoryMethodDescriptorRegistryWithIngressService now.