Repository: consensys/tessera
Issue: Fatal node sync call stops subsequent calls

When the PartyInfo poller makes requests to other nodes there are two classes of exceptions that can occur, general connection issues and then other, more serious issues.

A simple connection issue gets logged and passed over, as this is expected to happen from time to time, and all other exception types were presumed to be ephemeral and eventually resolve. We throw this error back up the call stack to handle later.

This means that when this error occurs, all other parties we wanted to sync with after the current do not happen (as the exception stops the loop). Sometimes, the error will never resolve, which effectively stops the node fully syncing with the network.

Any error, whether it is a general connection issue or something more serious, should not stop the processing of other calls.

Changed files:
tessera-partyinfo/src/main/java/com/quorum/tessera/partyinfo/PartyInfoPoller.java
tessera-core/src/main/java/com/quorum/tessera/threading/TesseraScheduledExecutor.java
tessera-partyinfo/src/test/java/com/quorum/tessera/partyinfo/PartyInfoPollerTest.java

diff --git a/tessera-core/src/main/java/com/quorum/tessera/threading/TesseraScheduledExecutor.java b/tessera-core/src/main/java/com/quorum/tessera/threading/TesseraScheduledExecutor.java
index 9bfc5579d..b46039041 100644
--- a/tessera-core/src/main/java/com/quorum/tessera/threading/TesseraScheduledExecutor.java
+++ b/tessera-core/src/main/java/com/quorum/tessera/threading/TesseraScheduledExecutor.java
@@ -10,10 +10,9 @@ import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
 
 /**
- * Schedules a continuous running task as an alternative to
- * a {@link Thread} running a {@code while(true)} loop
+ * Schedules a continuous running task as an alternative to a {@link Thread} running a {@code while(true)} loop
  *
- * Also allows delays if required between each execution of the loop
+ * <p>Also allows delays if required between each execution of the loop
  */
 public class TesseraScheduledExecutor {
 
@@ -24,13 +23,11 @@ public class TesseraScheduledExecutor {
     private final Runnable action;
 
     private final long rate;
-    
+
     private final long initialDelay;
 
-    public TesseraScheduledExecutor(final ScheduledExecutorService executor,
-                                    final Runnable action,
-                                    final long rate,
-                                    final long delay) {
+    public TesseraScheduledExecutor(
+            final ScheduledExecutorService executor, final Runnable action, final long rate, final long delay) {
         this.executor = Objects.requireNonNull(executor);
         this.action = Objects.requireNonNull(action);
         this.rate = rate;
@@ -38,21 +35,26 @@ public class TesseraScheduledExecutor {
     }
 
     /**
-     * Starts the submitted task and schedules it to run every given time frame
-     * Catches any Throwable and logs it so that the scheduling doesn't break
+     * Starts the submitted task and schedules it to run every given time frame. Catches any Throwable and logs it so
+     * that the scheduling doesn't break
      */
     @PostConstruct
     public void start() {
         LOGGER.info("Starting {}", this.action.getClass().getSimpleName());
 
-        final Runnable exceptionSafeRunnable = () -> {
-            try {
-                this.action.run();
-            } catch (final Throwable ex) {
-                LOGGER.error("Error when executing action {}", action.getClass().getSimpleName());
-                LOGGER.error("Error when executing action", ex);
-            }
-        };
+        final Runnable exceptionSafeRunnable =
+                () -> {
+                    try {
+                        LOGGER.debug("{} has started running", getClass().getSimpleName());
+
+                        this.action.run();
+                    } catch (final Throwable ex) {
+                        LOGGER.error("Error when executing action {}", action.getClass().getSimpleName());
+                        LOGGER.error("Error when executing action", ex);
+                    } finally {
+                        LOGGER.debug("{} has finished running", getClass().getSimpleName());
+                    }
+                };
 
         this.executor.scheduleWithFixedDelay(exceptionSafeRunnable, initialDelay, rate, TimeUnit.MILLISECONDS);
 
@@ -60,8 +62,8 @@ public class TesseraScheduledExecutor {
     }
 
     /**
-     * Stops any more executions of the submitted task from running
-     * Does not cancel the currently running task, which may be blocking
+     * Stops any more executions of the submitted task from running. Does not cancel the currently running task, which
+     * may be blocking
      */
     @PreDestroy
     public void stop() {
@@ -71,5 +73,4 @@ public class TesseraScheduledExecutor {
 
         LOGGER.info("Stopped {}", this.action.getClass().getSimpleName());
     }
-
 }
diff --git a/tessera-partyinfo/src/main/java/com/quorum/tessera/partyinfo/PartyInfoPoller.java b/tessera-partyinfo/src/main/java/com/quorum/tessera/partyinfo/PartyInfoPoller.java
index 14787ec66..35f45250e 100644
--- a/tessera-partyinfo/src/main/java/com/quorum/tessera/partyinfo/PartyInfoPoller.java
+++ b/tessera-partyinfo/src/main/java/com/quorum/tessera/partyinfo/PartyInfoPoller.java
@@ -5,11 +5,10 @@ import com.quorum.tessera.partyinfo.model.PartyInfo;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import java.net.ConnectException;
 import java.util.Objects;
 
 /**
- * Polls every so often to all known nodes for any new discoverable nodes This keeps all nodes up-to date and
+ * Polls every so often to all known nodes for any new discoverable nodes. This keeps all nodes up-to date and
  * discoverable by other nodes
  */
 public class PartyInfoPoller implements Runnable {
@@ -22,7 +21,7 @@ public class PartyInfoPoller implements Runnable {
 
     private final P2pClient p2pClient;
 
-    public PartyInfoPoller(PartyInfoService partyInfoService, P2pClient p2pClient) {
+    public PartyInfoPoller(final PartyInfoService partyInfoService, final P2pClient p2pClient) {
         this(partyInfoService, PartyInfoParser.create(), p2pClient);
     }
 
@@ -47,18 +46,15 @@ public class PartyInfoPoller implements Runnable {
      */
     @Override
     public void run() {
-        LOGGER.debug("Polling {}", getClass().getSimpleName());
-
         final PartyInfo partyInfo = partyInfoService.getPartyInfo();
-
         final byte[] encodedPartyInfo = partyInfoParser.to(partyInfo);
 
+        final String ourUrl = partyInfo.getUrl();
+
         partyInfo.getParties().stream()
-                .filter(party -> !party.getUrl().equals(partyInfo.getUrl()))
                 .map(Party::getUrl)
+                .filter(url -> !ourUrl.equals(url))
                 .forEach(url -> pollSingleParty(url, encodedPartyInfo));
-
-        LOGGER.debug("Polled {}. PartyInfo : {}", getClass().getSimpleName(), partyInfo);
     }
 
     /**
@@ -68,19 +64,11 @@ public class PartyInfoPoller implements Runnable {
      * @param encodedPartyInfo the encoded current party information
      */
     private void pollSingleParty(final String url, final byte[] encodedPartyInfo) {
-
         try {
             p2pClient.sendPartyInfo(url, encodedPartyInfo);
         } catch (final Exception ex) {
-
-            if (ConnectException.class.isInstance(ex.getCause())) {
-                LOGGER.warn("Server error {} when connecting to {}", ex.getMessage(), url);
-                LOGGER.debug(null, ex);
-            } else {
-                LOGGER.error("Error {} while executing poller for {}. ", ex.getMessage(), url);
-                LOGGER.debug(null, ex);
-                throw ex;
-            }
+            LOGGER.warn("Error {} when connecting to {}", ex.getMessage(), url);
+            LOGGER.debug(null, ex);
         }
     }
 }
diff --git a/tessera-partyinfo/src/test/java/com/quorum/tessera/partyinfo/PartyInfoPollerTest.java b/tessera-partyinfo/src/test/java/com/quorum/tessera/partyinfo/PartyInfoPollerTest.java
index 65d244e1c..484573b5e 100644
--- a/tessera-partyinfo/src/test/java/com/quorum/tessera/partyinfo/PartyInfoPollerTest.java
+++ b/tessera-partyinfo/src/test/java/com/quorum/tessera/partyinfo/PartyInfoPollerTest.java
@@ -6,7 +6,9 @@ import org.junit.After;
 import org.junit.Before;
 import org.junit.Test;
 
-import java.net.ConnectException;
+import java.util.Arrays;
+import java.util.HashSet;
+import java.util.Set;
 
 import static java.util.Collections.emptySet;
 import static java.util.Collections.singleton;
@@ -20,6 +22,8 @@ public class PartyInfoPollerTest {
 
     private static final String TARGET_URL = "http://bogus.com:9878/";
 
+    private static final String TARGET_URL_2 = "http://otherwebsite.com:9878/";
+
     private static final byte[] DATA = "BOGUS".getBytes();
 
     private PartyInfoService partyInfoService;
@@ -35,6 +39,9 @@ public class PartyInfoPollerTest {
         this.partyInfoService = mock(PartyInfoService.class);
         this.partyInfoParser = mock(PartyInfoParser.class);
         this.p2pClient = mock(P2pClient.class);
+
+        when(partyInfoParser.to(any(PartyInfo.class))).thenReturn(DATA);
+
         this.partyInfoPoller = new PartyInfoPoller(partyInfoService, partyInfoParser, p2pClient);
     }
 
@@ -45,109 +52,43 @@ public class PartyInfoPollerTest {
 
     @Test
     public void run() {
-
-        doReturn(true).when(p2pClient).sendPartyInfo(TARGET_URL, DATA);
-
         final PartyInfo partyInfo = new PartyInfo(OWN_URL, emptySet(), singleton(new Party(TARGET_URL)));
         doReturn(partyInfo).when(partyInfoService).getPartyInfo();
-
-        doReturn(DATA).when(partyInfoParser).to(partyInfo);
-
-        final PartyInfo updatedPartyInfo = new PartyInfo(OWN_URL, emptySet(), singleton(new Party(TARGET_URL)));
-        doReturn(updatedPartyInfo).when(partyInfoParser).from(DATA);
+        doReturn(true).when(p2pClient).sendPartyInfo(TARGET_URL, DATA);
 
         partyInfoPoller.run();
 
         verify(partyInfoService).getPartyInfo();
-
         verify(partyInfoParser).to(partyInfo);
-
         verify(p2pClient).sendPartyInfo(TARGET_URL, DATA);
     }
 
     @Test
     public void testWhenURLIsOwn() {
-
-        doReturn(true).when(p2pClient).sendPartyInfo(OWN_URL, DATA);
-
         final PartyInfo partyInfo = new PartyInfo(OWN_URL, emptySet(), singleton(new Party(OWN_URL)));
         doReturn(partyInfo).when(partyInfoService).getPartyInfo();
         doReturn(DATA).when(partyInfoParser).to(partyInfo);
-
-        final PartyInfo updatedPartyInfo = mock(PartyInfo.class);
-        doReturn(updatedPartyInfo).when(partyInfoParser).from(DATA);
-
-        partyInfoPoller.run();
-
-        verify(partyInfoParser).to(partyInfo);
-        verify(partyInfoService).getPartyInfo();
-    }
-
-    @Test
-    public void testWhenPostFails() {
-
-        doReturn(false).when(p2pClient).sendPartyInfo(TARGET_URL, DATA);
-
-        final PartyInfo partyInfo = new PartyInfo(OWN_URL, emptySet(), singleton(new Party(TARGET_URL)));
-
-        doReturn(partyInfo).when(partyInfoService).getPartyInfo();
-        doReturn(DATA).when(partyInfoParser).to(partyInfo);
-
-        final PartyInfo updatedPartyInfo = mock(PartyInfo.class);
-        doReturn(updatedPartyInfo).when(partyInfoParser).from(DATA);
+        doReturn(true).when(p2pClient).sendPartyInfo(OWN_URL, DATA);
 
         partyInfoPoller.run();
 
-        verify(partyInfoParser, never()).from(DATA);
         verify(partyInfoParser).to(partyInfo);
         verify(partyInfoService).getPartyInfo();
-        verify(p2pClient).sendPartyInfo(TARGET_URL, DATA);
     }
 
     @Test
-    public void runThrowsException() {
-
-        final PartyInfo partyInfo = new PartyInfo(OWN_URL, emptySet(), singleton(new Party(TARGET_URL)));
-
+    public void exceptionThrowByPostDoesntBubble() {
+        final Set<Party> parties = new HashSet<>(Arrays.asList(new Party(TARGET_URL), new Party(TARGET_URL_2)));
+        final PartyInfo partyInfo = new PartyInfo(OWN_URL, emptySet(), parties);
         doReturn(partyInfo).when(partyInfoService).getPartyInfo();
-        doReturn(DATA).when(partyInfoParser).to(partyInfo);
-
-        PartyInfo updatedPartyInfo = mock(PartyInfo.class);
-        doReturn(updatedPartyInfo).when(partyInfoParser).from(DATA);
-
         doThrow(UnsupportedOperationException.class).when(p2pClient).sendPartyInfo(TARGET_URL, DATA);
 
         final Throwable throwable = catchThrowable(partyInfoPoller::run);
-        assertThat(throwable).isInstanceOf(UnsupportedOperationException.class);
-
-        verify(p2pClient).sendPartyInfo(TARGET_URL, DATA);
-
-        verify(partyInfoService).getPartyInfo();
-        verify(partyInfoService, never()).updatePartyInfo(updatedPartyInfo);
-        verify(partyInfoParser, never()).from(DATA);
-        verify(partyInfoParser).to(partyInfo);
-    }
-
-    @Test
-    public void runThrowsConnectionExceptionAndDoesNotThrow() {
-
-        final PartyInfo partyInfo = new PartyInfo(OWN_URL, emptySet(), singleton(new Party(TARGET_URL)));
-
-        doReturn(partyInfo).when(partyInfoService).getPartyInfo();
-        doReturn(DATA).when(partyInfoParser).to(partyInfo);
-
-        final PartyInfo updatedPartyInfo = mock(PartyInfo.class);
-        doReturn(updatedPartyInfo).when(partyInfoParser).from(DATA);
-
-        final RuntimeException connectionException = new RuntimeException(new ConnectException("OUCH"));
-        doThrow(connectionException).when(p2pClient).sendPartyInfo(TARGET_URL, DATA);
-
-        partyInfoPoller.run();
 
+        assertThat(throwable).isNull();
         verify(p2pClient).sendPartyInfo(TARGET_URL, DATA);
-
+        verify(p2pClient).sendPartyInfo(TARGET_URL_2, DATA);
         verify(partyInfoService).getPartyInfo();
-        verify(partyInfoParser, never()).from(DATA);
         verify(partyInfoParser).to(partyInfo);
     }