feedback

cmd/process.go

@@ -81,8 +81,17 @@ func Start(a *AppState) *cobra.Command {
 				go c.TrackLatestBlockHeight(cmd.Context(), logger)
 
 				// wait until height is available
-				for c.LatestBlock() == 0 {
-					time.Sleep(1 * time.Second)
+				maxRetries := 45
+				for i := 0; i < maxRetries; i++ {
+					if c.LatestBlock() == 0 {
+						time.Sleep(1 * time.Second)
+					} else {
+						break
+					}
+					if i == maxRetries-1 {
+						logger.Error("Unable to get height")
+						os.Exit(1)
+					}
 				}
 
 				if err := c.InitializeBroadcaster(cmd.Context(), logger, sequenceMap); err != nil {
@@ -106,13 +115,14 @@ func Start(a *AppState) *cobra.Command {
 				go StartProcessor(cmd.Context(), a, registeredDomains, processingQueue, sequenceMap)
 			}
 
+			defer func() {
+				for _, c := range registeredDomains {
+					fmt.Printf("\n%s: latest-block: %d last-flushed-block: %d", c.Name(), c.LatestBlock(), c.LastFlushedBlock())
+					c.CloseClients()
+				}
+			}()
+
 			<-cmd.Context().Done()
-			// clean up
-			time.Sleep(20 * time.Millisecond)
-			for _, c := range registeredDomains {
-				fmt.Printf("\n%s: latest-block: %d last-flushed-block: %d", c.Name(), c.LatestBlock(), c.LastFlushedBlock())
-				c.CloseClients()
-			}
 		},
 	}
 

ethereum/listener.go

@@ -147,13 +147,15 @@ func (e *Ethereum) getAndConsumeHistory(
 	var history []ethtypes.Log
 	var err error
 
+	if start > end {
+		logger.Error(fmt.Sprintf("Unable to get history from %d to %d where the start block is greater than the end block", start, end))
+		return
+	}
+
 	// handle historical queries in chunks (some websockets only allow small history queries)
-	chunkSize := uint64(100)
+	const chunkSize = uint64(100)
 	chunk := 1
-	totalChunksNeeded := (end - start) / chunkSize
-	if (end-start)%chunkSize > 0 || totalChunksNeeded == 0 {
-		totalChunksNeeded++
-	}
+	totalChunksNeeded := (end - start + chunkSize - 1) / chunkSize
 
 	for start < end {
 		fromBlock := start