[Feat.Req] TAP Agent gRPC API

[Jannis]

Problem statement

For facilitating payments for Firehose and Substreams data, we will need to extend the functionality of TAP agent as described in this issue. Note: The bold parts in the use case description are what require new TAP agent functionality.

This assumes that indexers will run e.g. a Firehose Indexer Service that consumers will open a connection with in order to exchange TAP receipts and RAVs. This service will need to know when to request an RAV from a consumer and when to stop serving them.

Unlike with gateways that have a public endpoint to send RAV requests to, the only way to get RAVs from consumers is by "talking" to them directly. We therefore envision roughly the following architecture for this:

flowchart TB

  subgraph Indexer
    fis[Firehose Indexer Service]
    f[Firehose]
  end

  subgraph Consumer
    fc[Firehose Client]
    fnc[Firehose Network Client]
  end

  fc -- Firehose requests --> fnc
  fnc -- Data stream --> fc
  fis -- Authorization response, receipt & RAV requests --> fnc
  fnc -- Authorization request, receipts & RAVs --> fis

  fnc -- Firehose requests --> f
  f -- Data stream --> fnc
  f -- Report bytes sent/read --> fis

How does this work in practice?

1. Firehose Client wants to stream some data and makes requests to a local Firehose Network Client.
2. Firehose Network Client picks an indexer to use and opens a payment connection with its Firehose Indexer Service.
3. Firehose Indexer Service decides whether it still owes the consumer some data from a previous receipt.
   1. If yes, it sends an authorization message back that includes a Firehose URL and auth token.
   2. If no, it sends a receipt request back. Once it gets a receipt from the Firehose Network Client, it sends the authorization message.
4. Firehose Network Client opens the data connection and starts streaming data from the indexer's Firehose.
5. Firehose Indexer Service tracks the bytes served to the consumer against the latest receipt it has received.
   1. When it has served the data corresponding to the receipt amount, it requests another receipt.
6. Firehose Indexer Service also
   1. periodically checks how much collateral the consumer has remaining. If this ever goes to zero, it instructs the Firehose to terminate the data connection.
   2. periodically checks whether it needs a RAV from the consumer. When it does, it sends a RAV request to the consumer and waits for a RAV. If it doesn't get one back in a certain time frame, it instructs Firehose to terminate the data connection.

Proposal

We propose that TAP agent serves a gRPC API for Firehose Indexer Services but also Subgraph Indexer Services to connect to. This API could look as follows:

service TAPAgent {
  rpc PayerStatus(PayerStatusRequest) returns (PayerStatusResponse);
}

message PayerStatusRequest {
  bytes payer_address = 1;
}

message PayerStatusResponse {
  bytes payer_address = 1;
  bytes remaining_collateral = 2;
  optional bytes rav_request = 3;
}

Using this, different indexer service implementations can:

1. Decide when to stop serving a consumer (e.g. when their remaining collateral goes to zero).
2. Forward RAV requests to the consumer whenever necessary.

The subgraph indexer service could use this by periodically checking the payer status for all gateways it has interacted with. If a RAV is required for any of them, it could then send that request to their aggregator endpoint. This way, TAP agent would not need to know anything about gateways and their URLs.

The Firehose indexer service could use this by periodically checking the payer status for all consumers that have a payment connection open with the indexer. It can forward RAV requests to them via these payment connections.

Alternative considerations

• The above gRPC API may not be ideal for performance reasons. I (Jannis) am not a gRPC/protobuf expert. Perhaps a stream would be better instead of a request/response pattern? 🤔

Additional context

• It may make sense to also use this gRPC API between the subgraph indexer service and TAP agent. The way that could work is that the subgraph indexer service would be the one to have awareness of what gateways exist and what their RAV endpoints are. It could then periodically poll the TAP agent for whether it needs a RAV from any of the gateways, and take action accordingly. This way, the TAP agent would not have to be aware of gateways at all and the way Firehose indexer service, subgraph indexer service and others obtain RAV requests would be uniform.

[aasseman]

From @Jannis https://github.com/graphprotocol/indexer-rs/pull/83#discussion_r1380178322:

Once gateways register somewhere (they will do this for subscriptions — perhaps that's the contract you're referring to here or is there another one for TAP?), this could make for a nice Eventual in the indexer service. Add the TAP agent gRPC API, pipe the gateways eventual into an eventual that polls the "need a RAV?" status for each gateway periodically and whenever the gateways change, pipe that into code that requests the RAVs and stores them back in the DB. Done? 😁

[abourget]

We had discussed something like:

service {
  rpc ContinuityCheck(ContinuityRequest) ContinuityResponse;
  rpc SubmitReceipt(SubmitReceiptRequest) ContinuityResponse;
  rpc SubmitVoucher(SubmitReceiptRequest) ContinuityResponse; 
}

message ContinuityRequest {
  strong payer_address = 1;
}
message ContinuityResponse {
  uint64 remaining_collateral = 2; // on-chain collateral with all the vouchers and additional receipts I've received.
}
message SubmitReceiptRequest {
  bytes tap_payload_for_a_receipt = 1;
}
message SubmitVoucherRequest {
  bytes tap_payload_for_a_voucher = 1;
}

in our other post here: https://github.com/streamingfast/firehose-core/issues/18

We don't need to submit receipts? Is there another path that's already taken into account with a different protocol in the TAP agent already?

[Jannis]

@abourget Right now, the design is that there is a component that would live inside the Firehose Indexer Service that writes receipts and vouchers straight to the database, so the SubmitReceipt and SubmitVoucher messages aren't necessary. I know you had concerns about having multiple processes write to the receipt/voucher database, but perhaps we can try it this way first.

My thinking now is the following:

• We add the above gRPC API (open to whether we make it a stream or just an RPC method, thoughts?) to TAP agent.
• We then (or in parallel) implement the Firehose Network Client and Firehose Indexer Service.

My recommendation would be to write the latter two in Rust, since we can then use existing functionality to create receipts, discover indexers in the network (both on the client side) as well as handle receipts, store them in the database etc. (client side). I can put some documentation together for what these pieces are that we can use for this.

[abourget]

I would assume that he who owns the component chooses how and with what language to build it.

Having the database interactions behind the TAP agent would be a nice way to split responsibilities, and ownership.

In our original design, the voucher stuff was completely opaque and transitive to the indexer service. A better segregation of responsibilities I think.