A Typing Discipline for Statically Verified Crash Failure Handling in Distributed Systems

Distributed Programs, Partial Failures, and Coordination. Developing programs that execute across a set of physically remote, networked processes is challenging. The correct operation of a distributed program requires correctly designed protocols by which concurrent processes interact asynchronously, and correctly implemented processes according to their roles in the protocols. This becomes particularly challenging when distributed programs have to be resilient to partial failures, where some processes crashes while others remain operational. Partial failures affect both safety and liveness of applications. Asynchrony is the key issue, resulting in the inability to distinguish slow processes from failed ones. In general, this makes it impossible for processes to reach agreement, even when only a single process can crash [19].

In practice, such impasses are overcome by making appropriate assumptions for the considered infrastructure and applications. One common approach is to assume the presence of a highly available coordination service [26] – realized using a set of replicated processes large enough to survive common rates of process failures (e.g., 1 out of 3, 2 out of 5) – and delegating critical decisions to this service. While this coordinator model has been in widespread use for many years (cf. consensus service [22]), the advent of cloud computing has recently brought it further into the mainstream, via instances like Chubby [4] and ZooKeeper [26]. Such systems are used not only by end applications but also by a variety of frameworks and middleware systems across the layers of the protocol stack [11, 20, 31, 40].

Typing Disciplines for Distributed Programs. Typing disciplines for distributed programs is a promising and active research area towards addressing the challenges in the correct development of distributed programs. See Hüttel et al. [27] for a broad survey. Session types are one of the established typing disciplines for message passing systems. Originally developed in the \(\pi \)-calculus [23], these have been later successfully applied to a range of practical languages, e.g., Java [25, 41], Scala [39], Haskell [34, 38], and OCaml [28, 37]. Multiparty session types (MPSTs) [15, 24] generalize session types beyond two participants. In a nutshell, a standard MPST framework takes (1) a specification of the whole multiparty message protocol as a global type; from which (2) local types, describing the protocol from the perspective of each participant, are derived; these are in turn used to (3) statically type check the I/O actions of endpoint programs implementing the session participants. A well-typed system of session endpoint programs enjoys important safety and liveness properties, such as no reception errors (only expected messages are received) and session progress. A basic intuition behind MPSTs is that the design (i.e., restrictions) of the type language constitutes a class of distributed protocols for which these properties can be statically guaranteed by the type system.

Unfortunately, no MPST work supports protocols for asynchronous distributed programs dealing with partial failures due to process crashes, so the aforementioned properties no longer hold in such an event. Several MPST works have treated communication patterns based on exception messages (or interrupts) [6, 7, 16]. In these works, such messages may convey exceptional states in an application sense; from a protocol compliance perspective, however, these messages are the same as any other message communicated during a normal execution of the session. This is in contrast to process failures, which may invalidate already in-transit (orphan) messages, and where the task of agreeing on the concerted handling of a crash failure is itself prone to such failures.

Outside of session types and other type-based approaches, there have been a number of advances on verifying fault tolerant distributed protocols and applications (e.g., based on model checking [29], proof assistants [44]); however, little work exists on providing direct compile-time support for programming such applications in the spirit of MPSTs.

Contributions and Challenges. This paper puts forward a new typing discipline for safe specification and implementation of distributed programs prone to process crash failures based on MPSTs. The following summarizes the key challenges and contributions.    

To fit our model to practice, we introduce programming constructs similar to well-known and intuitive exception handling mechanisms, for handling concurrent and asynchronous process crash failures in sessions. These constructs serve to integrate user-level session control flow in endpoint processes and the underlying communications with the coordination service, used by the target applications of our work to outsource critical failure management decisions (see Fig. 1). It is important to note that the coordinator does not magically solve all problems. Key design challenges are to ensure that communication with it is fully asynchronous as in real-life, and that it is involved only in a “minimal” fashion. Thus we treat the coordinator as a first-class, asynchronous network artifact, as opposed to a convenient but impractical global “oracle” (cf. [6]), and our operational semantics of multiparty sessions remains primarily choreographic in the original spirit of distributed MPSTs, unlike works that resort to a centralized orchestrator to conduct all actions [5, 8]. As depicted in Fig. 1, application-specific communication does not involve the coordinator. Our model lends itself to common practical scenarios where processes monitor each other in a peer-based fashion to detect failures, and rely on a coordinator only to establish agreement on which processes have failed, and when.

A long version of this paper is available online [43]. The long version contains: full formal definitions, full proofs, and a prototype implementation in Scala.

Example. As a motivating example, Fig. 2 gives a global formal specification for a big data streaming task between a distributed file system (DFS) \( {dfs} \), and two workers \( w_{{1\!,2}} \). The DFS streams data to two workers, which process the data and write the result back. Most DFSs have built-in fault tolerance mechanisms [20], so we consider \( {dfs} \) to be robust, denoted by the annotation \([ \! \! [ {dfs} ] \! \! ]\); the workers, however, may individually fail. In the try-handle construct \(\textsf {t}(... )\textsf {h}( ... )\), the try-block \(\textsf {t}(...)\) gives the normal (i.e., failure-free) flow of the protocol, and \(\textsf {h(...)}\) contains the explicit handlers for potential crashes. In the try-block, the workers receive data from the DFS (\( {dfs} {\rightarrow } w_{i} {\ }{}\!\!\)), perform local computations, and send back the result (\( w_{i} {\rightarrow } {dfs} {\ }{}\!\!\)). If a worker crashes (\(\{{ w_{i} }\}\!:\!\ ...\)), the other worker will also take over the computation of the crashed worker, allowing the system to still produce a valid result. If both workers crash (by any interleaving of their concurrent crash events), the global type specifies that the DFS should safely terminate its role in the session.

We shall refer to this basic example, that focuses on the new failure handling constructs, in explanations in later sections. We also give many further examples throughout the following sections to illustrate the potential session errors due to failures exposed by our model, and how our framework resolves them to recover MPST safety and progress.

Roadmap. Section 2 describes the adopted system and failure model. Section 3 introduces global types for guiding failure handling. Section 4 introduces our process calculus with failure handling capabilities and a coordinator. Section 5 introduces local types, derived from global types by projection. Section 6 describes typing rules, and defines coherence of session environments with respect to endpoint crashes. Section 7 states properties of our model. Section 8 discusses related work. Sect. 9 draws conclusions.

Concretely we make the following assumptions on failures and the system:(1)–(3) are standard in literature on fault-tolerant distributed systems [19].

Note that processes can still recover but will not do so within sessions (or will not be re-considered for those). Other failure models, e.g., network partitions [21] or Byzantine failures [32], are subject of future work. The former are not tolerated by ZooKeeper et al., and the latter have often been argued to be a too generic failure model (e.g., [3]).

The assumption on the coordinator (4) implicitly means that the number of concomitant failures among the coordinator replicas is assumed to remain within a minority, and that failed replicas are replaced in time (to tolerate further failures). Without loss of validity, the coordinator internals can be treated as a blackbox. The final assumption (5) on failure detection is backed in practice by the concept of program-controlled crash [10], which consists in communicating decisions to disregard supposedly failed processes also to those processes, prompting them to reset themselves upon false suspicion. In practice systems can be configured to minimize the probability of such events, and by a “two-level” membership consisting in evicting processes from individual sessions (cf. recovery above) more quickly than from a system as a whole; several authors have also proposed network support to entirely avoid false suspicions (e.g., [33]).

These assumptions do not make handling of failures trivial, let alone mask them. For instance, the network can arbitrarily delay messages and thus reorder them with respect to their real sending times, and (so) different processes can detect failures at different points in time and in different orders.

Based on the foundations of MPSTs, we develop global types to formalize specifications of distributed protocols with explicit handling of partial failures due to role crashes, simply referred to as failures. We present global types before introducing the process calculus to provide a high-level intuition of how failure handling works in our model.

The syntax of global types is depicted in Fig. 3. We use the following base notations: \( p , q , ...\) for role (i.e., participant) names; \(l_1, l_2, ...\) for message labels; and \( t , t ', ...\) for type variables. Base types \(S\) may range over, \(\textsf {bool}, \textsf {int}\), etc.

Global types are denoted by G. We first summarize the constructs from standard MPST [15, 24]. A branch type \(p \rightarrow q \{l_i(S_i){.} G_i\}_{i \in I}\) means that \( p \) can send to \( q \) one of the messages of type \(S_k\) with label \(l_k\), where k is a member of the non-empty index set I. The protocol then proceeds according to the continuation \(G_k\). When I is a singleton, we may simply write \(p {\rightarrow } q {\ l(S)}.G\). We use \( t \) for type variables and take an equi-recursive view, i.e., \(\mu t {.} G\) and its unfolding \([\mu t {.} G / t ]\) are equivalent. We assume type variable occurrences are bound and guarded (e.g., \(\mu t {.} t \) is not permitted). \(\textsf {end}\) is for termination.

We now introduce our extensions for partial failure handling. A try-handle \(\textsf {t}(G_1 )\textsf {h}( H )^{\kappa }{.}G_2\) describes a “failure-atomic” protocol unit: all live (i.e., non-crashed) roles will eventually reach a consistent protocol state, despite any concurrent and asynchronous role crashes. The try-block \(G_1\) defines the default protocol flow, and \( H \) is a handling environment. Each element of \( H \) maps a handler signature \(F\), that specifies a set of failed roles \(\{ p _i\}_{i \in I}\), to a handler body specified by a G. The handler body G specifies how the live roles should proceed given the failure of roles \(F\). The protocol then proceeds (for live roles) according to the continuation \(G_2\) after the default block \(G_1\) or failure handling defined in \( H \) has been completed as appropriate.

To simplify later technical developments, we annotate each try-handle term in a given G by a unique \(\kappa \in \mathbb {N}\) that lexically identifies the term within G. These annotations may be assigned mechanically. As a short hand, we refer to the try-block and handling environment of a particular try-handle by its annotation; e.g., we use \(\kappa \) to stand for \(\textsf {t}(G_1 )\textsf {h}( H )^{\kappa }\). In the running examples (e.g., Fig. 2), if there exists only one try-handle, we omit \(\kappa \) for simplicity.

Top-Level Global Types and Robust Roles. We use the term top-level global type to mean the source protocol specified by a user, following a typical top-down interpretation of MPST frameworks [15, 24]. We allow top-level global types to be optionally annotated \([ \! \! [ \tilde{ p } ] \! \! ]G\), where \([ \! \! [ \tilde{ p } ] \! \! ]\) specifies a set of robust roles—i.e., roles that can be assumed to never fail. In practice, a participant may be robust if it is replicated or is made inherently fault tolerant by other means (e.g., the participant that represents the distributed file system in Fig. 2).

Well-Formedness. The first stage of validation in standard MPSTs is to check that the top-level global type satisfies the supporting criteria used to ensure the desired properties of the type system. We first list basic syntactic conditions which we assume on any given G: (i) each \(F\) is non-empty; (ii) a role in a \(F\) cannot occur in the corresponding handler body (a failed role cannot be involved in the handling of its own failure); and (iii) every occurrence of a non-robust role \( p \) must be contained within a, possibly outer, try-handle that has a handler signature \(\{ p \}\) (the protocol must be able to handle its potential failure). Lastly, to simplify the presentation without loss of generality, we impose that separate branch types not defined in the same default block or handler body must have disjoint label sets. This can be implicitly achieved by combining label names with try-handle annotations.

Assuming the above, we define well-formedness for our extended global types. We write \(G' \in G\) to mean that \(G'\) syntactically occurs in G (\(\in \) is reflexive); similarly for the variations \(\kappa \in G\) and \(\kappa \in \kappa '\). Recall \(\kappa \) is shorthand for \(\textsf {t}(G_1 )\textsf {h}( H )^\kappa \). We use a lookup function \( outer_{G} (\kappa )\) for the set of all try-handles in G that enclose a given \(\kappa \), including \(\kappa \) itself, defined by \( outer_{G} (\kappa ) =\{\kappa '\ |\ \kappa \in \kappa ' \wedge \kappa ' \in G \}\).

The first condition asserts that for any two separate handler signatures of a handling environment of \(\kappa \), there always exists a handler whose handler signature matches the union of their respective failure sets – this handler is either inside the handling environment of \(\kappa \) itself, or in the handling environment of an outer try-handle. This ensures that if roles are active in different handlers of the same try-handle then there is a handler whose signature corresponds to the union over the signatures of those different handlers. Example 2 together with Example 3 in Sect. 4 illustrate a case where this condition is needed. The second condition asserts that if the handling environment of a try-handle contains a handler for \(F\), then there is no outer try-handle with a handler for \(F'\) such that \(F' \subseteq F\). The reason for this condition is that in the case of nested try-handles, our communication model allows separate try-handles to start failure handling independently (the operational semantics will be detailed in the next section; see (TryHdl) in Fig. 6). The aim is to have the relevant roles eventually converge on performing the handling of the outermost try-handle, possibly by interrupting the handling of an inner try-handle. Consider the following example:

Figure 4 depicts a scenario that can occur in practical asynchronous systems with coordinator-based failure handling through frameworks such as ZooKeeper (Sect. 2). Using this scenario, we first illustrate challenges, formally define our model, and then develop a safe type system.

The scenario corresponds to a global type of the form \(\textsf {t}(G)\textsf {h}(\{P_a\}\!:\! G_a,\{P_a,P_d\}\!:\! G_{ad},...)^\kappa \), with processes \(P_{a..d}\) and a coordinator \( \varPsi \). We define a task to mean a unit of interactions, which includes failure handling behaviors. Initially all processes are collaborating on a task \(\phi \), which we label \((\kappa , \emptyset )\) (identifying the task context, and the set of failed processes). The shaded boxes signify which task each process is working on. Dotted arrows represent notifications between processes and \( \varPsi \) related to task completion, and solid arrows for failure notifications from \( \varPsi \) to processes. During the scenario, \(P_a\) first fails, then \(P_d\) fails: the execution proceeds through failure handling for \(\{P_a\}\) and \(\{P_a,P_d\}\).

Processes. Figure 5 defines the grammar of processes and (distributed) applications. Expressions \(e, e_i, ..\) can be values \(v, v_i, ...\), variables \(x, x_i, ...\), and standard operations. (Application) processes are denoted by \(P, P_i, ...\). An initialization \(a[p](y){.} P\) agrees to play role \(p\) via shared name a and takes actions defined in \(P\); actions are executed on a session channel \(\textsf {c}: \eta \), where \(\textsf {c}\) ranges over \(\textsf {s}[ p ]\) (session name and role name) and session variables y; \(\eta \) represents action statements.

A try-handle \(\textsf {t}(\eta )\textsf {h}( \mathsf {H} )^{\phi }\) attempts to execute the local action \(\eta \), and can handle failures occurring therein as defined in the handling environment \(\mathsf {H}\), analogously to global types. \(\mathsf {H}\) thus also maps a handler signature \(F{}\) to a handler body \(\eta \) defining how to handle \(F\). Annotation \(\phi = (\kappa , F)\) is composed of two elements: an identity \(\kappa \) of a global try-handle, and an indication of the current handler signature which can be empty. \(F= \emptyset \) means that the default try-block is executing, whereas \(F\not = \emptyset \) means that the handler body for \(F\) is executing. Term \(\underline{0}\) only occurs in a try-handle during runtime. It denotes a yielding for a notification from a coordinator (introduced shortly).

Other statements are similar to those defined in [15, 24]. Term \(0\) represents an idle action. For convention, we omit \(0\) at the end of a statement. Action \( p ! \ l (e){.} \eta \) represents a sending action that sends \( p \) a label l with content e, then it continues as \(\eta \). Branching \( p ? \{ l_i (x_i){.} \eta _i\}_{i \in I}\) represents a receiving action from \( p \) with several possible branches. When label \(l_k\) is selected, the transmitted value v is saved in \(x_k\), and \(\eta _k \{ v / x_k \}\) continues. For convenience, when there is only one branch, the curly brackets are omitted, e.g., \(\textsf {c}: p ? l(x){.} P\) means there is only one branch l(x). \(X \langle e \rangle \) is for a statement variable with one parameter e, and \(\textsf {def} \ D \ \textsf {in} \ \eta \) is for recursion, where declaration \(D\) defines the recursive body that can be called in \(\eta \). The conditional statement is standard.

The structure of processes ensures that failure handling is not interleaved between different sessions. However, we note that in standard MPSTs [15, 24], session interleaving must anyway be prohibited for the basic progress property. Since our aim will be to show progress, we disallow session interleaving within process bodies. Our model does allow parallel sessions at the top-level, whose actions may be concurrently interleaved during execution.

(Distributed) Systems. A (distributed) system in our programming framework is a composition of an application, which contains more than one process, and a coordinator (cf. Fig. 1). A system can be running within a private session \(\textsf {s}\), represented by \(( \nu \textsf {s}) \mathcal {S}\), or \(\mathcal {S}\ | \ \mathcal {S}'\) for systems running in different sessions independently and in parallel (i.e., no session interleaving). The job of the coordinator is to ensure that even in the presence of failures there is consensus on whether all participants in a given try-handle completed their local actions, or whether failures need to be handled, and which ones. We use \( \varPsi = G: ({F}, d)\) to denote a (robust) coordinator for the global type G, which stores in \(({F}, d)\) the failures \(F\) that occurred in the application, and in \(d\) done notifications sent to the coordinator. The coordinator is denoted by \( \psi \) when viewed as a role.

A (distributed) application¹ is a process \(P\), a parallel composition \(N \! \mid \! N'\), or a global queue carrying messages \(\textsf {s}: h\). A global queue \(\textsf {s}: h\) carries a sequence of messages \(m\), sent by participants in session \(\textsf {s}\). A message is either a regular message \( \langle p , q , {l}(v) \rangle \) with label l and content v sent from \( p \) to \( q \) or a notification. A notification may contain the role of a coordinator. There are done and failure notifications with two kinds of done notifications \(dn\) used for coordination: \( \langle p , \psi \rangle ^{\phi }\) notifies \( \psi \) that \( p \) has finished its local actions of the try-handle \(\phi \); \( \langle \psi , p \rangle ^{\phi }\) is sent from \( \psi \) to notify \( p \) that \( \psi \) has received all done notifications for the try-handle \(\phi \) so that \( p \) shall end its current try-handle and move to its next task. For example, in Fig. 4 at time (4) the coordinator will inform \(P_b\) and \(P_c\) via \( \langle \psi , P_b \rangle ^{(\kappa ,\{{P_a,P_d}\})}{.} \langle \psi , P_c \rangle ^{(\kappa ,\{{P_a,P_d}\})}\) that they can finish the try-handle \((\kappa ,\{{P_a,P_d}\})\). Note that the appearance of \( \langle \psi , p \rangle ^{\phi }\) implies that the coordinator has been informed that all participants in \(\phi \) have completed their local actions. We define two kinds of failure notifications: \( \langle \! [ \psi , \textsf {crash }F ] \! \rangle \) notifies \( \psi \) that \(F\) occurred, e.g., \(\{{ q }\}\) means \( q \) has failed; is sent from \( \psi \) to notify \( p \) about the failure \(F\) for possible handling. We write short for ; similarly for \( \langle \psi , \widetilde{ p } \rangle ^{\phi }\). Following the tradition of other MPST works the global queue provides an abstraction for multiple FIFO queues, each queue being between two endpoints (cf. TCP) with no global ordering. Therefore \(m_i \cdot m_j\) can be permuted to \(m_j \cdot m_i\) in the global queue if the sender or the receiver differ. For example the following messages are permutable: \( \langle p , q , {l}(v) \rangle \cdot \langle p , q' , {l}(v) \rangle \) if \(q \ne q'\) and \( \langle p , q , {l}(v) \rangle \cdot \langle \psi , p \rangle ^{\phi }\) and . But is not permutable, both have the same sender and receiver (\( \psi \) is the sender of ).

Basic Dynamic Semantics for Applications. Figure 6 shows the operational semantics of applications. We use evaluation contexts as defined in Fig. 5. Context E is either a hole , a default context \(\textsf {t}(E )\textsf {h}( \mathsf {H} )^{\phi }{.} \eta \), or a recursion context \(\textsf {def} \ D \ \textsf {in} \ E\). We write \(E[\eta ]\) to denote the action statement obtained by filling the hole in \(E[\cdot ]\) with \(\eta \).

Rule (Link) says that (local) processes who agree on shared name a, obeying to some protocol (global type), playing certain roles \( p_i \) represented by \(a[ p_i ](y_i){.}P\), together will start a private session \(\textsf {s}\); this will result in replacing every variable \(y_i\) in \(P_i\) and, at the same time, creating a new global queue \(\textsf {s}: \emptyset \), and appointing a coordinator \(G: (\emptyset , \emptyset )\), which is novel in our work.

Rule (Snd) in Fig. 6 reduces a sending action \( q ! \ l (e)\) by emitting a message \( \langle p , q , l(v) \rangle \) to the global queue \(\textsf {s}: h\). Rule (Rcv) reduces a receiving action if the message arriving at its end is sent from the expected sender with an expected label. Rule (Rec) is for recursion. When the recursive body, defined inside \(\eta \), is called by \(X \langle e \rangle \) where e is evaluated to v, it reduces to the statement \(\eta \{ v / x \} \) which will again implement the recursive body. Rule (Str) says that processes which are structurally congruent have the same reduction. Processes, applications, and systems are considered modulo structural congruence, denoted by \(\equiv \), along with \(\alpha \)-renaming. Rule (Par) and (Str) together state that a parallel composition has a reduction if its sub-application can reduce. Rule (Sys) states that a system has a reduction if its application has a reduction, and (New) says a reduction can proceed under a session. Rule (Crash) states that a process on channel \(\textsf {s}[ p ]\) can fail at any point in time. (Crash) also adds a notification which is sent to \( \psi \) (the coordinator). This is an abstraction for the failure detector described in Sect. 2 (5), the notification is the first such notification issued by a participant based on its local failure detector. Adding the notification into the global queue instead of making the coordinator immediately aware of it models that failures are only detected eventually. Note that a failure is not annotated with a level because failures transcend all levels, and asynchrony makes it impossible to identify “where” exactly they occurred. As a failure is permanent it can affect multiple try-handles. The (Crash) rule does not apply to participants which are robust, i.e., that conceptually cannot fail (e.g., \( {dfs} \) in Fig. 2). Rule (Crash) removes channel \(\textsf {s}[ p ]\) (the failed process) from application N, and removes messages and notifications delivered from, or heading to, the failed \( p \) by function \(\textit{remove}(h, p )\). Function \(\textit{remove}(h, p )\) returns a new queue after removing all regular messages and notifications that contain \( p \), e.g., let . Messages are removed to model that in a real system send/receive does not constitute an atomic action.

Handling at Processes. Failure handling, defined in Fig. 7, is based on the observations that (i) a process that fails stays down, and (ii) multiple processes can fail. As a consequence a failure can trigger multiple failure handlers either because these handlers are in different (subsequent) try-handles or because of additional failures. Therefore a process needs to retain the information of who failed. For simplicity we do not model state at processes, but instead processes read but do not remove failure notifications from the global queue. We define \( Fset (h, p )\) to return the union of failures for which there are notifications heading to \( p \), i.e., , issued by the coordinator in queue \(h\) up to the first done notification heading to \( p \):

In short, if the global queue is \(\emptyset \), then naturally there are no failure notifications. If the global queue contains a failure notification sent from the coordinator, say , we collect the failure. If the global queue contains done notification \( \langle \psi , p \rangle ^{\phi }\) sent from the coordinator then all participants in \(\phi \) have finished their local actions, which implies that the try-handle \(\phi \) can be completed. Our failure handling semantics, (TryHdl), allows a try-handle \(\phi = (\kappa , F)\) to handle different failures or sets of failures by allowing a try-handle to switch between different handlers. \(F\) thus denotes the current set of handled failures. For simplicity we refer to this as the current(ly handled) failure set. This is a slight abuse of terminology, done for brevity, as obviously failures are only detected with a certain lag. The handling strategy for a process is to handle the—currently—largest set of failed processes that this process has been informed of and is able to handle. This largest set is calculated by \(\cup \{ A \ | \ A \in \textit{dom}(\mathsf {H}) \wedge F\subset A \subseteq Fset (h, p )\}\), that selects all failure sets which are larger than the current one (\(A \in \textit{dom}(\mathsf {H}) \wedge F\subset A\)) if they are also triggered by known failures (\(A \subseteq Fset (h, p )\)). Condition \(F'\!: \eta ' \in \mathsf {H}\) in (TryHdl) ensures that there exists a handler for \(F'\). The following example shows how (TryHdl) is applied to switch handlers.

In Sect. 3 we formally define well-formedness conditions, which guarantee that if there exist two handlers for two different handler signatures in a try-handle, then a handler exists for their union. The following example demonstrates why such a guarantee is needed.

(I) explains that a process finishing its default action (\(P_b\)) cannot leave its current try-handle (\(\kappa ,\emptyset )\) immediately because other participants may fail (\(P_a\) failed). Below Eq. 1 also shows this issue from the perspective of semantics:In Eq. 1 the process acting on \(\textsf {s}[ p ]\) ended its try-handle (i.e., the action is \(0\) in the try-block), and if \(\textsf {s}[ p ]\) finishes its try-handle the participant acting on \(\textsf {s}[ q ]\) which started handling \(F\) would be stuck.

To solve the issue, we use (SndDone) and (RcvDone) for completing a local try-handle with the help of a coordinator. The rule (SndDone) sends out a done notification \( \langle p , \psi \rangle ^{\phi }\) if the current action in \(\phi \) is \(0\) and sets the action to \(\underline{0}\), indicating that a done notification from the coordinator is needed for ending the try-handle.

Assume process on channel \(\textsf {s}[ p ]\) finished its local actions in the try-block (i.e., as in Eq. 1 above), then by (SndDone), we havewhere notification \( \langle p , \psi \rangle ^{(\kappa , \emptyset )}\) is added to inform the coordinator. Now the process on channel \(\textsf {s}[ p ]\) can still handle failures defined in its handling environment. This is similar to the case described in (II).

Rule (RcvDone) is the counterpart of (SndDone). Once a process receives a done notification for \(\phi \) from the coordinator it can finish the try-handle \(\phi \) and reduces to the continuation \(\eta \). Consider Eq. 2 below, which is similar to Eq. 1 but we take a case where the try-handle can be reduced with (RcvDone). In Eq. 2 (SndDone) is applied:With \(h= \langle \psi , q \rangle ^{(\kappa , \emptyset )} \cdot \langle \psi , p \rangle ^{(\kappa , \emptyset )} \cdot \langle \! [ q , \textsf {crash }F ] \! \rangle \cdot \langle \! [ p , \textsf {crash }F ] \! \rangle \) both processes can apply (RcvDone) and safely terminate the try-handle \((\kappa , \emptyset )\). Note that \( Fset (h, p ) =\ Fset (h, q ) = \emptyset \) (by Definition 2), i.e., rule (TryHdl) can not be applied since a done notification suppresses the failure notification. Thus Eq. 2 will reduce to:It is possible that \(\eta '\) or \(\eta ''\) have handlers for \(F\). Note that once a queue contains \( \langle \psi , p \rangle ^{(\kappa , \emptyset )}\), all non-failed process in the try-handle \((\kappa , \emptyset )\) have sent done notifications to \( \psi \) (i.e. applied rule (SndDone)). The coordinator which will be introduced shortly ensures this.

Rule (Cln) removes a normal message from the queue if the label in the message does not exist in the target process, which can happen when a failure handler was triggered. The function \( labels ({\eta })\) returns all labels of receiving actions in \(\eta \) which are able to receive messages now or possible later. This removal based on the syntactic process is safe because in a global type separate branch types not defined in the same default block or handler body must have disjoint sets of labels (c.f., Sect. 3). Let \(\phi \in P\) if try-handle \(\phi \) appears inside P. Rule (ClnDone) removes a done notification of \(\phi \) from the queue if no try-handle \(\phi \) exists, which can happen in case of nesting when a handler of an outer try-handle is triggered.

Handling at Coordinator. Figure 8 defines the semantics of the coordinator. We firstly give the auxiliary definition of \(\textit{roles}(G)\) which gives the set of all roles appearing in G.

In rule \(({\mathbf {\mathsf{{F}}}})\), \({F}\) represents the failures that the coordinator is aware of. This rule states that the coordinator collects and removes a failure notification heading to it, retains this notification by \(G: ( {F}', {d} )\), \({F}' = {F}\cup \{ p \}\), and issues failure notifications to all non-failed participants.

Rules (CollectDone, IssueDone), in short inform all participants in \(\phi = (\kappa , F)\) to finish their try-handle \(\phi \) if the coordinator has received sufficient done notifications of \({\phi }\) and did not send out failure notifications that interrupt the task \((\kappa , F)\) (e.g. see (III)). Rule (CollectDone) collects done notifications, i.e., \( \langle p , \psi \rangle ^{\phi }\), from the queue and retains these notification; they are used in (IssueDone). For introducing (IssueDone), we first introduce \( hdl (G, (\kappa , F))\) to return a set of handler signatures which can be triggered with respect to the current handler:

Also, we abuse the function \(\textit{roles}\) to collect the non-coordinator roles of \(\phi \) in \(d\), written \(\textit{roles}(d, \phi )\); similarly, we write \(\textit{roles}(G, \phi )\) where \(\phi = (\kappa , F)\) to collect the roles appearing in the handler body \(F\) in the try-handle of \(\kappa \) in G. Remember that \(d\) only contains done notifications sent by participants.

Rule (IssueDone) is applied for some \(\phi \) when conditions \(\forall F' \in hdl (G, \phi ){.}(F' \not \subseteq {F})\) and \(\textit{roles}(d, \phi ) \supseteq \textit{roles}(G,\phi ) \setminus {{F}}\) are both satisfied, where \({F}\) contains all failures the coordinator is aware of. Intuitively, these two conditions ensure that (1) the coordinator only issues done notifications to the participants in the try-handle \(\phi \) if it did not send failure notifications which will trigger a handler of the try-handle \(\phi \); (2) the coordinator has received all done notifications from all non-failed participants of \(\phi \). We further explain both conditions in the following examples, starting from condition \(\forall F' \in hdl (G, \phi ){.}(F' \not \subseteq {F})\), which ensures no handler in \(\phi \) can be triggered based on the failure notifications \({F}\) sent out by the coordinator.

Another condition \(\textit{roles}(d, \phi ) \supseteq \textit{roles}(G,\phi ) \setminus {F}\) states that only when the coordinator sees sufficient done notifications (in \(d\)) for \(\phi \), it issues done notifications to all non-failed participants in \(\phi \), i.e., \( \langle \psi , \textit{roles}(G,\phi ) \setminus {F} \rangle ^{\phi }\). Recall that \(\textit{roles}(d, \phi )\) returns all roles which have sent a done notification for the handling of \(\phi \) and \(\textit{roles}(G,\phi )\) returns all roles involving in the handling of \(\phi \). Intuitively one might expect the condition to be \(\textit{roles}(d, \phi ) = \textit{roles}(G,\phi )\); the following example shows why this would be wrong.

The interplay between issuing of done notification (IssueDone) and issuing of failure notifications \(({\mathbf {\mathsf{{F}}}})\) is non-trivial. The following proposition clarifies that the participants in the same try-handle \(\phi \) will never get confused with handling failures or completing the try-handle \(\phi \).

Figure 9 defines local types for typing behaviors of endpoint processes with failure handling. Type \( p ! \) is the primitive for a sending type, and \( p ? \) is the primitive for a receiving type, derived from global type \(p \rightarrow q \{l_i(S_i){.} G_i\}_{i \in I}\) by projection. Others correspond straightforwardly to process terms. Note that type \(\underline{\textsf {end}}\) only appears in runtime type checking. Below we define \( G \! \! \upharpoonright _{} \! \! p \) to project a global type G on \( p \), thus generating \( p \)’s local type.

The main rule is (1): if \( p \) appears somewhere in the target try-handle global type then the endpoint type has a try-handle annotated with \(\kappa \) and the default logic (i.e., \({F}= \emptyset \)). Note that even if \( G_0 \! \! \upharpoonright _{} \! \! p = \textsf {end}\) the endpoint still gets such a try-handle because it needs to be ready for (possible) failure handling; if \( p \) does not appear anywhere in the target try-handle global type, then the projection skips to the continuation.

Rule (2) produces local types for interaction endpoints. If the endpoint is a sender (i.e., \( p = p _1\)), then its local type abstracts that it will send something from one of the possible internal choices defined in \(\{ l_i (S_i)\}_{i \in I}\) to \( p _2\), then continue as \( G_k \! \! \upharpoonright _{} \! \! p \), gained from the projection, if \(k \in I\) is chosen. If the endpoint is a receiver (i.e., \( p = p _2\)), then its local type abstracts that it will receive something from one of the possible external choices defined in \(\{ l_i (S_i)\}_{i \in I}\) sent by \( p _1\); the rest is similarly as for the sender. However, if \( p \) is not in this interaction, then its local type starts from the next interaction which \( p \) is in; moreover, because \( p \) does not know what choice that \( p _1\) has made, every path \( G_i \! \! \upharpoonright _{} \! \! p \) lead by branch \(l_i\) shall be the same for \( p \) to ensure that interactions are consistent. For example, in \(G = p _1 \rightarrow p _2 \{ l_1(S_1){.} p _3 \rightarrow p _1\ l_3(S), \ l_2(S_2){.} p _3 \rightarrow p _1\ l_4(S) \}\), interaction \( p _3 \rightarrow p _1\) continues after \( p _1 \rightarrow p _2\) takes place. If \(l_3 \not = l_4\), then G is not projectable for \( p _3\) because \( p _3\) does not know which branch that \( p _1\) has chosen; if \( p _1\) chose branch \(l_1\), but \( p _3\) (blindly) sends out label \(l_4\) to \( p _1\), for \( p _1\) it is a mistake (but it is not a mistake for \( p _3\)) because \( p _1\) is expecting to receive label \(l_3\). To prevent such inconsistencies, we adopt the projection algorithm proposed in [24]. Other session type works [17, 39] provide ways to weaken the classical restriction on projection of branching which we use.

Rule (3) forbids a try-handle to appear in a recursive body, e.g., \(\mu t {.} \textsf {t}(G )\textsf {h}( F\!:\! t )^\kappa {.}G\) is not allowed, but \(\textsf {t}(\mu t {.} G )\textsf {h}( H )^\kappa \) and \(\textsf {t}(G )\textsf {h}( F\!:\! \mu t {.} G', H )^\kappa \) are allowed. This is because \(\kappa \) is used to avoid confusion of messages from different try-handles. If a recursive body contains a try-handle, we have to dynamically generate different levels to maintain interaction consistency, so static type checking does not suffice. We are investigating alternative runtime checking mechanisms, but this is beyond the scope of this paper. Other rules are straightforward.

Next we introduce our type system for typing processes. Figures 10 and 11 present typing rules for endpoints processes, and typing judgments for applications and systems respectively.

We define shared environments \(\varGamma \) to keep information on variables and the coordinator, and session environments \(\varDelta \) to keep information on endpoint types:\(\varGamma \) maps process variables X and content variables x to their types, shared names a to global types G, and a coordinator \( \varPsi = G: ({F}, d)\) to failures and done notifications it has observed. \(\varDelta \) maps session channels \(\textsf {c}\) to local types and session queues to queue types. We write \(\varGamma , \varGamma ' = \varGamma \cup \varGamma '\) when \(\textit{dom}(\varGamma ) \cap \textit{dom}(\varGamma ') = \emptyset \); same for \(\varDelta , \varDelta '\). Queue types \(\mathtt {h}\) are composed of message types \(\mathtt {m}\). Their permutation is defined analogously to the permutation for messages. The typing judgment for local processes \(\varGamma \vdash \ P \rhd \ \varDelta \) states that process \(P\) is well-typed by \(\varDelta \) under \(\varGamma \).

Since we do not define sequential composition for processes, our type system implicitly forbids session interleaving by . This is different from other session type works [15, 24], where session interleaving is prohibited for the progress property; here the restriction is inherent to the type system.

Figure 10 lists our typing rules for endpoint processes. Rule says that if a process’s set of actions is well-typed by \( G \! \! \upharpoonright _{} \! \! p \) on some \(\textsf {c}\), this process can play role \( p \) in a, which claims to have interactions obeying behaviors defined in G. \( \langle G \rangle \) means that G is closed, i.e., devoid of type variables. This rule forbids \(a[ p ]{.} b[ q ]{.} P\) because a process can only use one session channel. Rule states that an action for sending is well-typed to a sending type if the label and the type of the content are expected; states that an action for branching (i.e., for receiving) is well-typed to a branching type if all labels and the types of contents are as expected. Their follow-up actions shall also be well-typed. Rule types an idle process. Predicate end-only \(\varDelta \) is defined as stating whether all endpoints in \(\varDelta \) have type \(\textsf {end}\):

Rule types yielding actions, which only appear at runtime. Rule is standard in the sense that the process is well-typed by \(\varDelta \) if e has boolean type and its sub-processes (i.e., \(\eta _1\) and \(\eta _2\)) are well-typed by \(\varDelta \). Rules are based on a recent summary of MPSTs [14]. Note that forbids the type \(\mu t {.} t \). Rule states that a try-handle is well-typed if it is annotated with the expected level \(\phi \), its default statement is well-typed, \(\mathcal {H}\) and \(\mathsf {H}\) have the same handler signatures, and all handling actions are well-typed.

Figure 11 shows typing rules for applications and systems. Rule types an empty queue. Rules simply type messages based on their shapes. Rule says two applications composed in parallel are well-typed if they do not share any session channel. Rule says a part of a system \(\mathcal {S}\) can start a private session, say \(\textsf {s}\), if \(\mathcal {S}\) is well-typed according to a \(\varGamma \vdash \ \varDelta _{\textsf {s}}\) that is coherent (defined shortly). The system \((\nu \textsf {s})\mathcal {S}\) with a part becoming private in \(\textsf {s}\) is well-typed to \(\varDelta \setminus \varDelta _{\textsf {s}}\), that is, \(\varDelta \) after removing \(\varDelta _{\textsf {s}}\).

Rule says that a system is well-typed if application N is well-typed and there exists a coordinator \( \varPsi \) for handling this application. We say \(\varGamma \vdash \ \varDelta \) is coherent under \(\varGamma \) if the local types of all endpoints are dual to each other after their local types are updated because of messages or notifications in \(s: \mathtt {h}\).

Coherence. We say that a session environment is coherent if, at any time, given a session with its latest messages and notifications, every endpoint participating in it is able to find someone to interact with (i.e., its dual party exists) right now or afterwards.

We write \(\textsf {s}[ p ]: T \bowtie \textsf {s}[ q ]: T'\) to state that actions of the two types are dual:

Operation \( T \! \downharpoonright \! p \) is to filter T to get the partial type which only contains actions of \( p \). For example, \( p_1 ! l' (S' ){.} p_2 ! l(S) \! \downharpoonright \! p _2 = p_2 ! l(S)\) and \( p_1 ! \{ T_1, T_2\} \! \downharpoonright \! p _2 = p_2 ? l(S)\) where \(T_1 = l_1(S_1){.} p_2 ? l(S)\) and \(T_2 = l_2(S_2){.} p_2 ? l(S)\). Next we define \( (\mathtt {h})_{ p \rightarrow q }\) to filter \(\mathtt {h}\) to generate (1) the normal message types sent from \( p \) heading to \( q \), and (2) the notifications heading to \( q \). For example . The message types are abbreviated to contain only necessary information.

We define \( T {-} \mathtt {ht}\) to mean the effect of \(\mathtt {ht}\) on T. Its concept is similar to the session remainder defined in [35], which returns new local types of participants after participants consume messages from the global queue. Since failure notifications will not be consumed in our system, and we only have to observe the change of a participant’s type after receiving or being triggered by some message types in \(\mathtt {ht}\), we say that \( T {-} \mathtt {ht}\) represents the effect of \(\mathtt {ht}\) on T. The behaviors follows our operational semantics of applications and systems defined in Figs. 6, 7, and 8. For example \( \textsf {t}( q ? \{ l_i(S_i) {.} T_i \}_{i \in I} )\textsf {h}( \mathcal {H} )^{\phi }{.}T' {-} q ? l_k (S_k){\cdot } \mathtt {ht} = \textsf {t}(T_k )\textsf {h}( \mathcal {H} )^{\phi }{.}T' {-} \mathtt {ht}\) where \(k \in I\).

Now we define what it means for \(\varDelta \) to be coherent under \(\varGamma \):

In condition 1, we require a coordinator for every session so that when a failure occurs, the coordinator can announce failure notifications to ask participants to handle the failure. Condition 2 requires that, for any two endpoints, say \(\textsf {s}[ p ]\) and \(\textsf {s}[ q ]\), in \(\varDelta \), equation \(\textsf {s}[ p ]: T \! \downharpoonright \! q {-} (\mathtt {h})_{ q \rightarrow p } \bowtie \textsf {s}[ q ]: T' \! \downharpoonright \! p {-} (\mathtt {h})_{ p \rightarrow q }\), must hold. This condition asserts that interactions of non-failed endpoints are dual to each other after the effect of \(\mathtt {h}\); while failed endpoints are removed from \(\varDelta \), thus the condition is satisfied immediately.

We show that our type system ensures properties of subject congruence, subject reduction, and progress. All auxiliary definitions and proofs are in the long version [43].

The property of subject congruence states that if \(\mathcal {S}\) (a system containing an application and a coordinator) is well-typed by some session environment, then a \(\mathcal {S}'\) that is structurally congruent to it is also well-typed by the same session environment:

Subject reduction states that a well-typed \(\mathcal {S}\) (coherent session environment respectively) is always well-typed (coherent respectively) after reduction:

We allow sessions to run in parallel at the top level, e.g., . Assume we have \(\mathcal {S}\) with \(a[ p ]{.} P\in \mathcal {S}\). If we cannot apply rule \(({\mathbf {\mathsf{{Link}}}})\), \(\mathcal {S}\) cannot reduce. To prevent this kind of situation, we require \(\mathcal {S}\) to be initializable such that, \(\forall a[ p ]{.} P\in \mathcal {S}\), \(({\mathbf {\mathsf{{Link}}}})\) is applicable.

The following property states that \(\mathcal {S}\) never gets stuck (property of progress):

After all processes in \(\mathcal {S}\) terminate, failure notifications sent by coordinators are left; thus the final system can be of the form , where \(h, ..., h'\) have failure notifications sent by coordinators and thus reduction rules \(({\mathbf {\mathsf{{\text {CollectDone}}}}}), ({\mathbf {\mathsf{{\text {IssueDone}}}}})\), and \(({\mathbf {\mathsf{{F}}}})\) will not be applied.

Minimality. The following proposition points out that, when all roles defined in a global type, say G, are robust, then the application obeying to G will never have interaction with a coordinator (i.e., interactions of the application are equivalent to those without a coordinator). This is an important property, as it states that our model does not incur coordination overhead when all participants are robust, or in failure-agnostic contexts as considered in previous MPST works.

Several session type works study exception handling [7, 9, 16, 30]. However, to the best of our knowledge this is the first theoretical work to develop a formalism and typing discipline for the coordinator-based model of crash failure handling in practical asynchronous distributed systems.

Structured interactional exceptions [7] study exception handling for binary sessions. The work extends session types with a try-catch construct and a throw instruction, allowing participants to raise runtime exceptions. Global escape [6] extends previous works on exception handling in binary session types to MPSTs. It supports nesting and sequencing of try-catch blocks with restrictions. Reduction rules for exception handling are of the form \(\varSigma \vdash P \rightarrow \varSigma ' \vdash P'\), where \(\varSigma \) is the exception environment. This central environment at the core of the semantics is updated synchronously and atomically. Furthermore, the reduction of a try-catch block to its continuation is done in a synchronous reduction step involving all participants in a block. Lastly this work can only handle exceptions, i.e., explicitly raised application-level failures. These do not affect communication channels [6], unlike participant crashes.

Similarly, our previous work [13] only deals with exceptions. An interaction \(p \rightarrow q : S \vee F\) defines that p can send a message of type S to q. If F is not empty then instead of sending a message p can throw F. If a failure is thrown only participants that have casual dependencies to that failure are involved in the failure handling. No concurrent failures are allowed therefore all interactions which can raise failures are executed in a lock step fashion. As a consequence, the model can not be used to deal with crash failures.

Adameit et al. [1] propose session types for link failures, which extend session types with an optional block which surrounds a process and contains default values. The default values are used if a link failure occurs. In contrast to our work, the communication model is overall synchronous whereas our model is asynchronous; the optional block returns default values in case of a failure but it is still the task of the developer to do something useful with it.

Demangeon et al. study interrupts in MPSTs [16]. This work introduces an interruptible block \(\{|G|\}^c \langle l\ \text {by}\ \mathtt {r}\rangle ;G'\) identified by c; here the protocol G can be interrupted by a message l from \(\mathtt {r}\) and is continued by \(G'\) after either a normal or an interrupted completion of G. Interrupts are more a control flow instruction like exceptions than an actual failure handling construct, and the semantics can not model participant crashes.

Neykova and Yoshida [36] show that MPSTs can be used to calculate safe global states for a safe recovery in Erlang’s let it crash model [2]. That work is well suited for recovery of lightweight processes in an actor setting. However, while it allows for elaborate failure handling by connecting (endpoint) processes with runtime monitors, the model does not address the fault tolerance of runtime monitors themselves. As monitors can be interacting in complex manners replication does not seem straightforwardly applicable, at least not without potentially hampering performance (just as with straightforward replication of entire applications).

Failure handling is studied in several process calculi and communication-centered programming languages without typing discipline. The conversation calculus [42] models exception behavior in abstract service-based systems with message-passing based communication. The work does not use channel types but studies the behavioral theory of bisimilarity. Error recovery is also studied in a concurrent object setting [45]; interacting objects are grouped into coordinated atomic actions (CAs) which enable safe error recovery. CAs can however not be nested. PSYNC [18] is a domain specific language based on the heard-of model of distributed computing [12]. Programs written in PSYNC are structured into rounds which are executed in a lock step manner. PSYNC comes with a state-based verification engine which enables checking of safety and liveness properties; for that programmers have to define non-trivial inductive invariants and ranking functions. In contrast to the coordinator model, the heard-of model is not widely deployed in practice. Verdi [44] is a framework for implementing and verifying distributed systems in Coq. It provides the possibility to verify the system against different network models. Verdi enables the verification of properties in an idealized fault model and then transfers the guarantees to more realistic fault models by applying transformation functions. Verdi supports safety properties but no liveness properties.