VDB-1219: handle eof errors (#59)

* Don't return UnexpectedEOF when extracting logs

* Remove logging redundant logs in EventWatcher

libraries/shared/watcher/event_watcher.go

@@ -17,6 +17,7 @@
 package watcher
 
 import (
+	"io"
 	"time"
 
 	"github.com/makerdao/vulcanizedb/libraries/shared/constants"
@@ -92,14 +93,16 @@ func (watcher *EventWatcher) Execute(recheckHeaders constants.TransformerExecuti
 
 func (watcher *EventWatcher) extractLogs(recheckHeaders constants.TransformerExecution, errs chan error, quitChan chan bool) {
 	call := func() error { return watcher.LogExtractor.ExtractLogs(recheckHeaders) }
-	watcher.withRetry(call, watcher.ExpectedExtractorError, "extracting", errs, quitChan)
+	// io.ErrUnexpectedEOF errors are sometimes returned from fetching logs at the head of the chain when fetching from an uncle or fork block
+	expectedErrors := []error{watcher.ExpectedExtractorError, io.ErrUnexpectedEOF}
+	watcher.withRetry(call, expectedErrors, "extracting", errs, quitChan)
 }
 
 func (watcher *EventWatcher) delegateLogs(errs chan error, quitChan chan bool) {
-	watcher.withRetry(watcher.LogDelegator.DelegateLogs, watcher.ExpectedDelegatorError, "delegating", errs, quitChan)
+	watcher.withRetry(watcher.LogDelegator.DelegateLogs, []error{watcher.ExpectedDelegatorError}, "delegating", errs, quitChan)
 }
 
-func (watcher *EventWatcher) withRetry(call func() error, expectedErr error, operation string, errs chan error, quitChan chan bool) {
+func (watcher *EventWatcher) withRetry(call func() error, expectedErrors []error, operation string, errs chan error, quitChan chan bool) {
 	defer close(errs)
 	consecutiveUnexpectedErrCount := 0
 	for {
@@ -111,9 +114,8 @@ func (watcher *EventWatcher) withRetry(call func() error, expectedErr error, ope
 			if err == nil {
 				consecutiveUnexpectedErrCount = 0
 			} else {
-				if err != expectedErr {
+				if isUnexpectedError(err, expectedErrors) {
 					consecutiveUnexpectedErrCount++
-					logrus.Errorf("error %s logs: %s", operation, err.Error())
 					if consecutiveUnexpectedErrCount > watcher.MaxConsecutiveUnexpectedErrs {
 						errs <- err
 						return
@@ -124,3 +126,13 @@ func (watcher *EventWatcher) withRetry(call func() error, expectedErr error, ope
 		}
 	}
 }
+
+func isUnexpectedError(currentError error, expectedErrors []error) bool {
+	for _, expectedError := range expectedErrors {
+		if currentError == expectedError {
+			return false
+		}
+	}
+
+	return true
+}

libraries/shared/watcher/event_watcher_test.go

@@ -18,6 +18,7 @@ package watcher_test
 
 import (
 	"errors"
+	"io"
 	"time"
 
 	"github.com/makerdao/vulcanizedb/libraries/shared/constants"
@@ -127,6 +128,14 @@ var _ = Describe("Event Watcher", func() {
 			Expect(err).To(MatchError(errExecuteClosed))
 		})
 
+		It("does not treat an io.ErrUnexpectedEOF error from the node as an unexpected error", func() {
+			extractor.ExtractLogsErrors = []error{io.ErrUnexpectedEOF, errExecuteClosed}
+
+			err := eventWatcher.Execute(constants.HeaderUnchecked)
+
+			Expect(err).To(MatchError(errExecuteClosed))
+		})
+
 		It("extracts watched logs again if missing headers found", func() {
 			extractor.ExtractLogsErrors = []error{nil, errExecuteClosed}