Marlowe Runtime "Follower" Architecture

[jhbertra]

I've put together a high-level and rough diagram of the possible components of the "follower" replacement in the Marlowe Runtime. Here is the diagram:

flowchart
    node[MARLOWE_RT_NODE_SOCKET]
    sport[MARLOWE_RT_STREAMING_PORT]
    qport[MARLOWE_RT_QUERY_PORT]
    subgraph Marlowe Runtime
        cs[Chain Sync Client]
        hp[History Processor]
        hp[History Processor]
        hstore[History Store]
        cstore[Contract Store]
        sserv[Streaming Server]
        sser[Streaming API]
        qserv[Query Server]
        qser[Query API]
        cs--"Marlowe Chain Events"-->hp
        hp--History Events-->hstore
        hp--History Events-->cstore
        hp--History Events--->sserv
        hstore-.History Events.->sserv
        hstore-.History Events.->qserv
        cstore-.Contracts.->qserv
        sserv--Pub Messages-->sser
        qserv-.Queries/Results.->qser
    end
    node--Chain Events-->cs
    qser-.Protocol TBD.->qport
    sser--Protocol TBD-->sport

Loading

Interpreting the diagram:

• The arrows flow in the direction that data propagates.
• A solid line indicates a push-based connection.
• A dotted line indicates a pull-based connection.
• In a push-based connection, the upstream component pushes data through the channel to the downstream component, which handles it.
• A pull-based connection is a function call. The downstream component calls a function and the upstream component returns a result.

Here is a summary of the components:

Chain Sync Client

This component connects to the local node and handles the stream of events that it sends. Its role is to classify the raw transactions and decide if they are relevant to the Marlowe Runtime. Conceptually, it can be thought of as a function:

ChainEvent -> Maybe MarloweChainEvent

Where ChainEvent can be a roll forward or roll backward message, and MarloweChainEvent is a filtered version of the ChainEvent that contains only the information we care about.

The chain sync client pushes the resulting Marlowe Chain Events to its outbox.

History Processor

This component receives a stream of Marlowe Chain Events and converts them to History Events. Conceptually, it can be thought of as a function:

MarloweChainEvent -> [HistoryEvent]

The history processor pushes the resulting History Events to its outbox.

History Store:

This component receives history events and exposes an API for querying them.

Contract Store:

This component receives history events and aggregates them into contract state aggregate roots. It exposes an API for querying contract state.

Streaming Server:

This component serves streaming content from the runtime. It is responsible for establishing streaming connections in response to client requests. When it receives a subscription, it fast forwards the subscriber by querying the history store for the history "so far". It then switches to the live updates from the History Processor, sending them out to the subscriber.

Query Server:

This component responds to queries. It queries the History Store and the contract store to do so.

Streaming API:

This component connects the Streaming Server to a concrete transport protocol (e.g. websockets via HTTP, ZeroMQ, etc.)

Query API:

This component connects the Query Server to a concrete transport protocol (e.g. HTTP, ZeroMQ, RPC, GraphQL, etc.)

MARLOWE_RT_NODE_SOCKET:

This is a reference / path to the local node's socket, held in the environment.

MARLOWE_RT_STREAMING_PORT:

This is a port number for the Streaming API to bind to, held in the environment.

MARLOWE_RT_QUERY_PORT:

This is a port number for the Query API to bind to, held in the environment.

[bwbush]

This looks great, and I think it would be feasible and efficient. Here are two questions about eliminating redundant paths in order to reduce the chances for receiving conflicting or out-of-sync data.

1. Is the direct connection History Processor --> Streaming Server really necessary? I guess I'm wondering if fresh events and replay of events could be unified so that this connection wasn't necessary, and the Streaming Server would just receive events from History Store.
2. The two paths leading into Query Server could have synchronization problems, so we might need a simple coordination mechanism between them, or a least report (SlotNo, Hash Block) in each query result, so the UI can know if it's been fed out-of-sync information.

[jhbertra]

Hey Brian, thanks for the questions. Here are my thoughts:

1. That's a good point. First of all, let's assume each component is a single-threaded message processing loop. Then the only order I can see being a problem is this:
   1, History Processor sends event X for contract A to Streaming Server, which streaming server drops because it's not subscribed to contract A.
   2. Streaming Server receives a new subscription to contract A.
   3. Streaming Server queries History Store for history from contract A (which hasn't been told about event X)
   4. History Store responds to the query from Sreaming Server
   5. History Processor sends event X to History Store

   The result is that the new subscription won't ever see event X. Of course, this would require steps 2 and 3 to both preempt step 5. In practice, this seems incredibly unlikely, bot not impossible. Maybe it would be better to design something that avoids this pitfall altogether.

   To explore the other possible orders of events:

   • 1 2 3 4 5 Has the same problem (DB query will occur in step 3, before History Store gets event X)
   • 1 2 4 3 5 works because the query will return event X
   • 1 4 2 3 5, 4 1 2 3 5 both work for the same reason
   • 2 1 3 4 5 works assuming the Streaming Server remembers the events it sees while it is pulling the initial history
   • 2 3 1 4 5, 2 3 1 5 4, 2 3 5 1 4 all work for the same reason
   • 2 3 4 1 5 and 2 3 4 5 1 are the "happy path"
   • 2 3 1 4 5 works for the same reason

   An alternate configuration that avoids these problems is to forward all events through History Store as you suggest. The downside is that now History Store has two responsibilities, making it a greater liability (it starts to become a centralized point of failure). However, what we get for this is a consistency guarantee. We could formalize this guarantee with the following pseudocode invariant:

   subscribe historyStore \event -> do
     history <- query historyStore event.marloweParams
     assert $ contains event history

2. I am not as concerned about this as long as we design things to keep this in mind. There is a small chance you could query a contract and then query its history, and get an out-of-date contract (because the ContractStore has taken longer to process an event than the HistoryStore). But if you pulled again within a few milliseconds, you should see this sort its self out (the system is eventually consistent). One thing we could do is tag the contract state with the ID of the last event, and if a client queries the history and the contract and sees the contract is out-of-date, it could just re-query the contract with exponential falloff.

[jhbertra]

Also, regarding 1 we could just push all the logic involved with setting up a stream to History Store. Basically, History Store would expose two endpoints (expressed with an erlang-style actor model in mind):

-- | Subscribe the given process to history evens from the History Store that match the given HistoryFilter.
-- The subscriber is guaranteed to receive all matching events currently in the store in a single message before it receives
-- any new events, and the next event it receives will be the next sequential matching event (none will be skipped / dropped).
subscribe
  :: HistoryStoreRef -- ^ Reference to the history store
  -> ProcessRef -- ^ The process (in Erlang terms, not necessarily an OS process) to deliver messages to
  -> HistoryFilter -- ^ a filter for the stream.
  -> IO SubscriptionRef -- ^ reference to the subscription (used to unsubscribe)

-- | Query the History Store for history results matching the given filter. The query results will be delivered to the
-- | requesting (current) process' mailbox when they are available.
query
  :: HistoryStoreRef -- ^ Reference to the history store
  -> HistoryFilter -- ^ a filter for the results.
  -> IO QueryId -- ^ an ID for the query (will be delivered alongside the query results in the process' mailbox)

[bwbush]

Thanks for the explanation. Just as a safety measure, could each message or stream include a field with the current slot number, block number, and block hash? That would open the option for a client to identify mismatches.

[jhbertra]

@bwbush here is an updated diagram based on your first suggestion:

flowchart
    node[MARLOWE_RT_NODE_SOCKET]
    sport[MARLOWE_RT_STREAMING_PORT]
    qport[MARLOWE_RT_QUERY_PORT]
    subgraph Marlowe Runtime
        cs[Chain Sync Client]
        hp[History Processor]
        hp[History Processor]
        hstore[History Store]
        cstore[Contract Store]
        sserv[Streaming Server]
        sser[Streaming API]
        qserv[Query Server]
        qser[Query API]
        cs--"Marlowe Chain Events"-->hp
        hp--History Events-->hstore
        hp--History Events-->cstore
        hstore--History Events-->sserv
        hstore-.History Events.->qserv
        cstore-.Contracts.->qserv
        sserv--Pub Messages-->sser
        qserv-.Queries/Results.->qser
    end
    node--Chain Events-->cs
    qser-.Protocol TBD.->qport
    sser--Protocol TBD-->sport

Loading