Go: Distibuted ingestion

[bartekn]

@tamirms reminded about distributed ingestion in #1519. Basically for availability we need to have multiple ingesting instances, so when one instance goes down, the other one can continue. This is actually simple to do, we can use key-value store to block on the ledger number (SELECT ... FOR UPDATE) when ingestion is starting. Then we can have many idle ingestion processes that can continue when the main instance stops running.

However, the problem appears when there are multiple ingesting nodes with some in-memory stores (like order book graphs) that should be synchronized in a way that they always respond with a data for the same ledger.

I think in general we have the following options:

• Use master-standby pattern. Basically we have n ingesting nodes but only one is a master node that accepts all requests to in-memory structures. It's Horizon admin responsibility to automate flipping process to standby instance when master stops working. This is the simplest solution but require a few devops hours on Horizon admin side for initial configuration. @brahman81 from devops perspective do you think it is hard to do?
• Always keep the state for the last few (one?) ledgers. When one of the ingesting nodes is behind, it serves stale data. This was the idea we originally had to tackle this problem at the design meetings.
• Inter-node synchronization protocol. In short, all read requests are blocked until all ingesting nodes broadcast that the latest ledger has been processed by them. It probably requires another server to be source of truth to decide what to do if one of the instances stop working or is slow, maybe leader-election algorithm? It's probably the hardest solution but require no work on Horizon admin part (except defining IPs of the other nodes).

Any other ideas? @tamirms @ire-and-curses

[tomquisel]

Is there any need to do this for performance / load spreading reasons, or is it just availability?

If we're just going for availability, master-standby (with all nodes actively ingesting) seems like the simplest solution by far.

Another simple option is using ip-based affinity for load balancing. That means each client will get consistent results except possibly around the time of a server entering or exiting the rotation. It doesn't solve the issue of consistent results across clients though.

[brahman81]

Basically we have n ingesting nodes but only one is a master node that accepts all requests to in-memory structures

Routing in-memory structure requests to a specific node is fairly simple to setup but am not sure it's such a good idea from a scaling perspective. This node effectively becomes a bottleneck/single point of failure. Tools such as Kubernetes, Autoscaling groups or application load balancers can mitigate this to a certain degree if we decide to go down this road but it ends up making distributed Horizon deployments more complex imo.

[bartekn]

Is there any need to do this for performance / load spreading reasons, or is it just availability?

It's just for availability.

Another simple option is using ip-based affinity for load balancing. That means each client will get consistent results except possibly around the time of a server entering or exiting the rotation. It doesn't solve the issue of consistent results across clients though.

The problem with this is it's usually based on cookies so if the given client app doesn't properly implement the cookie-jar it may get invalid results. For ip-based affinity, I think some ISPs change their clients IPs from time to time so it may also not be fully reliable.

We just had a quick chat about this with @ire-and-curses and @tamirms and we decided that probably the best solution is to go with master-standby solution. The reasoning is as follows:

• The code solution in Horizon seems not trivial (leader election, handling timeouts, discoverability) and can cause a lot of bugs that are hard to track. It seems that products like ZooKeeper are built exactly to solve this issue and they are separate projects on their own.
• We don't have exact data what percentage of Horizon users actually need this. There were also no requests like this so far.
• _If_ a given organisation's use-case require a more complex deployment, they probably already have an ops team that can prepare master-standby architecture.

However I've exchanged a couple of messages with @brahman81 and he brought some valid points:

• Requests to a given endpoint routed to a single instance may slow it down, so it may be required to up the specs.
• In general, it adds a complexity compared to the current simple load-balancer.

We still have a couple weeks before the next release date so maybe we can explore the solution that squash offers at the single price level to see if it's faster? If it's not fast enough, maybe we can keep the graph on all nodes as @tamirms suggested. The up to date data is not a huge requirement in this case because there's always a time delay between showing results to the user and them picking an offer that works for them. And it seems that for other existing Horizon features in-memory data structure won't be needed.

[tomquisel]

Here's another simple alternative that I think you proposed before @bartekn:

• We allow users to pass in a min_ledger parameter to API calls we want to protect from consistency issues
• We return a current_ledger with API responses
• If a Horizon gets a request for a ledger that's larger than its current ledger it does one of two things:
  
  
  
  • if the min_ledger is 2 or less ledgers away from its current ledger, it blocks until min_ledger == current_ledger, or for a max of 20 secs.
  
  
  • otherwise, Horizon returns a 404.
  
  

It means a user can avoid ever seeing an older result after a newer result if they pass the current_ledger from their last request in with their current request as min_ledger. And it allows us to support full round robin load balancing.

[bartekn]

Let me propose another idea that is kind of mix of above.

First of all, some observations:

1. It seems that even selecting a few instances as ingesting requires some configuration that, even if is easy, can be error-prone.
2. The consistency of Horizon responses makes sense only when we have request batching or something like GraphQL. Without it, we can't really maintain consistent view for two requests. But that's great that we're thinking about it now.
3. If we decide to update the memory state of non-ingesting instances via DB it requires us to duplicate a lot of code and much more testing. We can end up with invalid state if the code doesn't work exactly the same as in ingesting node.

Here's the idea. We make all nodes ingesting (you can't make a node non-ingesting). This removes all configuration/routing/leader election issues. As of now, we have two stores: memory (OB graph) and Postgres DB (rest of the data). We make Postgres our single source of truth. Then the ingestion algorithm on each node is roughly:

1. SELECT ... last_ingested_ledger ... FOR UPDATE. This will make only one Horizon instance to actually get the value and the rest will be in a blocked state until a DB transaction commits/rollbacks. This is actually selecting a leader/_master_ randomly. See: Explicit locking.
2. We compare last_ingested_ledger value to the last processed ledger sequence (internal_last_ingested_ledger) processed by the current node.
3. If last_ingested_ledger == internal_last_ingested_ledger: the current node is the _master_ ingesting node in this round. We process the full pipeline.
4. Else: the other node processed this ledger, we omit processors that store results in a DB but do process data we keep in memory (OB graph in our case).

We can observe that all nodes other than _master_ will be a little bit behind in terms of updating memory data. How to solve this:

In action/controller that gets data from DB only we basically get the results, because DB results are always the latest ingested data.

In action/controller that gets data from memory:

• Simple solution: since we don't need consistency right now (observation 2) we just serve the graph we currently have in memory.
• Great solution: if we need consistency (we'll need it at some point) we have a couple options:
  
  
  
  • We fetch the last_ingested_ledger from DB and compare it to internal_last_ingested_ledger. If it's not equal, we wait for the current node to catchup (maybe using channel signal?).
  
  
  • We extract OB graph processing to another pipeline and keep the graph versions for last few ledger, we query the graph that matches the last_ingested_ledger. This exploits the fact that updating OB graph will probably always be faster that the pipeline modifying a DB.
  
  

Advantages:

• There's a single code that updates the state on all nodes (observation 3). However, some parts are turned off for _non-master_ ingesting node. It's easier than maintaining two versions of the code (for ingesting/non-ingesting node).
• Requests to the data is evenly distributed to all nodes (compared to only one ingesting node in the previously proposed solution).
• The configuration is easier.

Disadvantages:

• There will be some delays when requesting memory structures (for the simple version above).
• ???

[tamirms]

@bartekn I like the solution. My suggestion is that have two different ingestion pipelines, one for updating database state and another for updating in-memory data structures. The distributed lock will only be necessary for the database ingestion pipelines. The in-memory updating pipelines can operate independently without any synchronization.

If we need consistency for serving requests using in-memory data structures, I prefer the min_ledger approach outlined by @tomquisel .

The problem I see with the work flow of reading last_ingested_ledger from the db and then waiting until internal_last_ingested_ledger matches that value is that it's possible that the horizon instance serving the request catches up to last_ingested_ledger but by the time that happens the other master horizon instance already advanced to the next ledger.

[tamirms]

I also wanted to mention that, as an alternative to select for update, it's possible to implement distributed locks using advisory locks in postgres https://medium.com/@codeflows/application-level-locking-with-postgres-advisory-locks-c29759eeb7a7

[ire-and-curses]

The problem I see with the work flow of reading last_ingested_ledger from the db and then waiting until internal_last_ingested_ledger matches that value is that it's possible that the horizon instance serving the request catches up to last_ingested_ledger but by the time that happens the other master horizon instance already advanced to the next ledger.

@tamirms I agree with that analysis, but doesn't Tom's suggestion have the same possibility? During the 20s grace period before the server catches up or times out another server could advance to the next ledger.

[tamirms]

@ire-and-curses I think in tom's proposal we don't require that the response corresponds the latest ledger, we only require that the response corresponds to a ledger which is greater than or equal to the min_ledger parameter

[bartekn]

My suggestion is that have two different ingestion pipelines, one for updating database state and another for updating in-memory data structures. The distributed lock will only be necessary for the database ingestion pipelines. The in-memory updating pipelines can operate independently without any synchronization.

The problem is that LiveSession supports a single pipeline only. We can probably have another LiveSession. Let's discuss this during development.

If we need consistency for serving requests using in-memory data structures, I prefer the min_ledger approach outlined by @tomquisel .

It think the main problem with @tomquisel idea is that the average wait time will be long and it doesn't solve consistency in some cases. I need to make a request first to check what's the latest ledger and then try to guess what number to send to get a result (current_ledger+1/current_ledger+2). Possible cases are:

• [Optimistic] At the moment I send my request the min_ledger is actually equal current_ledger so I get the result right away.
• When I send my request and min_ledger>current_ledger: Then data from different stores can be at different ledgers min_ledger+X and min_ledger+X+1.
• When I send my request and min_ledger<current_ledger: I need to wait. But also it's possible that I get a signal to proceed only some stores are updated. It's not clear how to Tom wants to solve this.

Let me know if there's anything wrong in above.

The problem I see with the work flow of reading last_ingested_ledger from the db and then waiting until internal_last_ingested_ledger matches that value is that it's possible that the horizon instance serving the request catches up to last_ingested_ledger but by the time that happens the other master horizon instance already advanced to the next ledger.

It won't happen if we make all read queries to a DB in isolated transaction (REPEATABLE READ). In this isolation level the other changes (commited/uncommited) by other transactions are not seen in the current transaction. It can cause some problems if there are multiple writing transactions but in our case it will be just reading queries.

[tamirms]

The problem is that LiveSession supports a single pipeline only. We can probably have another LiveSession. Let's discuss this during development.

yes, I meant to say that there would be two sessions, one for the database updates and another for the in memory updates.

It think the main problem with @tomquisel idea is that the average wait time will be long and it doesn't solve consistency in some cases. I need to make a request first to check what's the latest ledger and then try to guess what number to send to get a result (current_ledger+1/current_ledger+2).

I think the point of the min_ledger approach is that we don't need all the horizon instances to be synchronized. I only care that the response is accurate for a given ledger.

I don't see why an API consumer would make a request to find the latest ledger before calling the path finding endpoint. In most cases, I would assume that API consumers would be content with the latest data available on the given horizon instance. Otherwise, if the response must match a specific ledger the ledger should be provided explicitly.

[bartekn]

I think the point of the min_ledger approach is that we don't need all the horizon instances to be synchronized. I only care that the response is accurate for a given ledger.

The problem is that data may not be consistent within a single node (memory vs database at different ledgers). Even when you pass min_ledger and wait, it's still possible, I explained how in https://github.com/stellar/go/issues/1529#issuecomment-514796637.

I don't see why an API consumer would make a request to find the latest ledger before calling the path finding endpoint. In most cases, I would assume that API consumers would be content with the latest data available on the given horizon instance.

So the min_ledger is an optional parameter?

Otherwise, if the response must match a specific ledger the ledger should be provided explicitly.

It's still not clear to me how this solution work. I guess some extra explanations would help. How does it work in the cases I listed above:

• [Optimistic] At the moment I send my request the min_ledger is actually equal current_ledger so I get the result right away.
• When I send my request and min_ledger>current_ledger: Then data from different stores can be at different ledgers min_ledger+X and min_ledger+X+1.
• When I send my request and min_ledger

[tamirms]

The problem is that data may not be consistent within a single node (memory vs database at different ledgers).

@bartekn , this is a very good point. I definitely missed this earlier and now I understand the rationale behind your approach. Given that requirement, I take back my earlier suggestions.

I think we should use the same session to process both the database ingestion and the order book graph as you originally proposed.

I have one question about the great solution, in the second bullet point you mention:

We extract OB graph processing to another pipeline and keep the graph versions for last few ledger, we query the graph that matches the last_ingested_ledger. This exploits the fact that updating OB graph will probably always be faster that the pipeline modifying a DB.

I thought that internal_last_ingested_ledger will always be less than or equal to last_ingested_ledger. There are two cases, either internal_last_ingested_ledger == last_ingested_ledger and we continue serving the path finding request with the in memory order book, or, internal_last_ingested_ledger < last_ingested_ledger and we block until the current node catches up. I don't see a scenario where we need to maintain several versions of the order book graph. Could you explain why it's necessary?

[edit: I reread the comment and now I understand the second bullet point was proposing an alternative solution to blocking until the current node catches up to last_ingested_ledger. Overall, the proposal makes sense to me. When we implement the simple solution, I think it would be good to add metrics to see how often internal_last_ingested_ledger is less than last_ingested_ledger when serving path finding requests. ]

[bartekn]

@tamir in the solution where we maintain two pipelines/sessions (one for OB
graph and the other one for DB updates), the OB pipeline will probably
always be faster. If the processing time of both pipelines is smaller than
average close time the difference should never be more than 1 ledger
(however can be more if there are delays). The reason we need OB graphs for
the last few ledgers is that DB state can be one (or more) ledger behind
the OB state. Here's an example sequence:

1. DB pipeline starts processing.

2. OB pipeline starts processing.

3. OB pipeline ends.

4. The user requests /paths.

5. We start a DB transaction with REPEATABLE READ isolation.

6. We query a DB to get the latest processed ledger (this is our source of
   truth). Let's say the returned value is X.

7. However OB graph represents X+1 (or X+2, X+3... in case of big delays).

8. DB pipeline ends and commits a transaction.

If we store OB graph for the last few ledgers, we can simply get the result
for the ledger X (instead of X+1). However, as I mentioned above we don't
need this until we have batching so the simple solution where the state of
OB graph is not in sync with a DB is fine.

W dniu czw., 25.07.2019 o 07:58 tamirms [e-mail]
napisał(a):

The problem is that data may not be consistent within a single node
(memory vs database at different ledgers).

@bartekn https://github.com/bartekn , this is a very good point. I
definitely missed this earlier and now I understand the rationale behind
your approach. Given that requirement, I take back my earlier suggestions.

I think we should use the same session to process both the database
ingestion and the order book graph as you originally proposed.

I have one question about the great solution, in the second bullet point
you mention:

We extract OB graph processing to another pipeline and keep the graph
versions for last few ledger, we query the graph that matches the
last_ingested_ledger. This exploits the fact that updating OB graph will
probably always be faster that the pipeline modifying a DB.

I thought that internal_last_ingested_ledger will always be less than or
equal to last_ingested_ledger. There are two cases, either internal_last_ingested_ledger
== last_ingested_ledger and we continue serving the path finding request
with the in memory order book, or, internal_last_ingested_ledger <
last_ingested_ledger and we block until the current node catches up. I
don't see a scenario where we need to maintain several versions of the
order book graph. Could you explain why it's necessary?

—
You are receiving this because you were mentioned.
Reply to this email directly, view it on GitHub
https://github.com/stellar/go/issues/1529?email_source=notifications&email_token=AADRQKQJHJAZ77GU2I5LOOTQBE6JDA5CNFSM4IF3IWW2YY3PNVWWK3TUL52HS4DFVREXG43VMVBW63LNMVXHJKTDN5WW2ZLOORPWSZGOD2YOBIQ#issuecomment-514908322,
or mute the thread
https://github.com/notifications/unsubscribe-auth/AADRQKWIHOBIFLIJLXLC6DDQBE6JDANCNFSM4IF3IWWQ
.

[tomquisel]

I like Bartek's great solution and would be happy with the simple solution for now.

Specifically, my favorite version of the great solution is this one:

We fetch the last_ingested_ledger from DB and compare it to internal_last_ingested_ledger. If it's not equal, we wait for the current node to catchup (maybe using channel signal?).

This seems simpler than maintaining multiple versions of the data in memory.

I think we should maintain slow consistency even without GraphQL or other request batching. This is the expectation that if you read a value from Horizon, you cannot then read an earlier value. I'd even go further, and say we should ensure that if you read a value from ledger N, you cannot then read a value from any ledger less than N.

I think the simple solution is fine for the very first version since ingestion is typically very fast and issues will rarely show up, but it'd be good to move to maintaining slow consistency in the short term.

I think Bartek's algorithm can be tweaked slightly (unless I misunderstood it) to minimize latency and also guarantee slow consistency.

Here's the tweaked version:

• The node fetches the last_ingested_ledger from DB at the time the request comes in. Call this L.
• We wait until internal_last_ingested_ledger >= L if needed.
• If more than some constant, say 10 seconds passes, we abort the request and return 500.
• Otherwise, we service the request.

What I don't love about this is that we're constantly fetching last_ingested_ledger from the DB on every request.

[tamirms]

What I don't love about this is that we're constantly fetching last_ingested_ledger from the DB on every request.

@tomquisel actually we only need to compare internal_last_ingested_ledger when handling requests which fetch data from the in memory orderbook. So, we'd only need to fetch last_ingested_ledger from the DB on path finding requests.

All other requests fetch their data from the DB and the data in the DB is consistent with last_ingested_ledger

[bartekn]

Closed in #1566.