From 58e00c105c25b4aae54a10147ee06449b10fbd88 Mon Sep 17 00:00:00 2001
From: Pekka Enberg <penberg@iki.fi>
Date: Fri, 2 Sep 2022 12:44:48 +0300
Subject: [PATCH] Fix lock re-entrancy in
 process_append_entry_request_as_follower()

In my project, I have a struct that implements both the `Cluster` and
`StateMachine` traits, and, therefore, also share the same `Mutex`
that's passed to `Replica`.

However, `process_append_entry_request_as_follower` is holding on to the
state machine lock while attempting to acquire the cluster lock, which
results in re-entrancy that causes a deadlock (as reported by
`no_deadlocks` crate):

```
A reentrance has been attempted, but `std::sync`'s locks are not reentrant. This results in a deadlock. dependence cycle: [Thread(ThreadId(11)), Lock(20)]
Lock taken at:
   0: little_raft::replica::Replica<C,M,T,D>::process_append_entry_request_as_follower
             at /Users/penberg/src/database/little-raft/little_raft/src/replica.rs:709:33
   1: little_raft::replica::Replica<C,M,T,D>::process_message_as_follower
             at /Users/penberg/src/database/little-raft/little_raft/src/replica.rs:766:18
   2: little_raft::replica::Replica<C,M,T,D>::process_message
             at /Users/penberg/src/database/little-raft/little_raft/src/replica.rs:352:32
   3: little_raft::replica::Replica<C,M,T,D>::poll_as_follower
             at /Users/penberg/src/database/little-raft/little_raft/src/replica.rs:332:21
   4: little_raft::replica::Replica<C,M,T,D>::start
             at /Users/penberg/src/database/little-raft/little_raft/src/replica.rs:233:36

Reentrace at:
   0: little_raft::replica::Replica<C,M,T,D>::process_append_entry_request_as_follower
             at /Users/penberg/src/database/little-raft/little_raft/src/replica.rs:735:27
   1: little_raft::replica::Replica<C,M,T,D>::process_message_as_follower
             at /Users/penberg/src/database/little-raft/little_raft/src/replica.rs:766:18
   2: little_raft::replica::Replica<C,M,T,D>::process_message
             at /Users/penberg/src/database/little-raft/little_raft/src/replica.rs:352:32
   3: little_raft::replica::Replica<C,M,T,D>::poll_as_follower
             at /Users/penberg/src/database/little-raft/little_raft/src/replica.rs:332:21
   4: little_raft::replica::Replica<C,M,T,D>::start
             at /Users/penberg/src/database/little-raft/little_raft/src/replica.rs:233:36
```

Fix the deadlock by moving entry processing to a separate function,
which reduces the scope of the state machine lock so that it is no
longer held when we attempt to acquire the cluster lock.

Fixes #42.
---
 little_raft/src/replica.rs | 43 +++++++++++++++++++++-----------------
 1 file changed, 24 insertions(+), 19 deletions(-)

diff --git a/little_raft/src/replica.rs b/little_raft/src/replica.rs
index c13219e..d400fe1 100644
--- a/little_raft/src/replica.rs
+++ b/little_raft/src/replica.rs
@@ -705,25 +705,7 @@ where
             return;
         }
 
-        let mut state_machine = self.state_machine.lock().unwrap();
-        for entry in entries {
-            // Drop local inconsistent logs.
-            if entry.index <= self.get_last_log_index()
-            && entry.term != self.get_term_at_index(entry.index).unwrap() {
-                for i in entry.index..self.log.len() {
-                    state_machine.register_transition_state(
-                        self.log[i].transition.get_id(),
-                        TransitionState::Abandoned(TransitionAbandonedReason::ConflictWithLeader)
-                    );
-                } 
-                self.log.truncate(entry.index);
-            }
-
-            // Push received logs.
-            if entry.index == self.log.len() + self.index_offset {
-                self.log.push(entry);
-            }
-        }
+        self.process_entries(entries);
 
         // Update local commit index to either the received commit index or the
         // latest local log position, whichever is smaller.
@@ -745,6 +727,29 @@ where
         );
     }
 
+    fn process_entries(&mut self, entries: Vec<LogEntry<T>>) {
+        let mut state_machine = self.state_machine.lock().unwrap();
+        for entry in entries {
+            // Drop local inconsistent logs.
+            if entry.index <= self.get_last_log_index()
+                && entry.term != self.get_term_at_index(entry.index).unwrap()
+            {
+                for i in entry.index..self.log.len() {
+                    state_machine.register_transition_state(
+                        self.log[i].transition.get_id(),
+                        TransitionState::Abandoned(TransitionAbandonedReason::ConflictWithLeader),
+                    );
+                }
+                self.log.truncate(entry.index);
+            }
+
+            // Push received logs.
+            if entry.index == self.log.len() + self.index_offset {
+                self.log.push(entry);
+            }
+        }
+    }
+
     fn process_message_as_follower(&mut self, message: Message<T, D>) {
         match message {
             Message::VoteRequest {