adjust chan buff sizes; move trace msg to trace logrus level

pkg/eth/streamer.go

@@ -25,7 +25,7 @@ import (
 )
 
 const (
-	PayloadChanBufferSize = 20000 // the max eth sub buffer size
+	PayloadChanBufferSize = 10000 // half the max eth sub buffer size
 )
 
 // StreamClient is an interface for subscribing and streaming from geth

pkg/eth/transformer.go

@@ -119,7 +119,7 @@ func (sdt *StateDiffTransformer) Transform(workerID int, payload statediff.Paylo
 			traceMsg += fmt.Sprintf("postgres transaction commit duration: %s\r\n", tDiff.String())
 		}
 		traceMsg += fmt.Sprintf(" TOTAL PROCESSING TIME: %s\r\n", time.Now().Sub(start).String())
-		logrus.Info(traceMsg)
+		logrus.Trace(traceMsg)
 	}()
 	tDiff = time.Now().Sub(t)
 	prom.SetTimeMetric("t_free_postgres", tDiff)

pkg/sync/service.go

@@ -31,10 +31,6 @@ import (
 	"github.com/vulcanize/ipld-eth-indexer/pkg/shared"
 )
 
-const (
-	PayloadChanBufferSize = 2000
-)
-
 // Indexer is the top level interface for streaming, converting to IPLDs, publishing, and indexing all chain data at head
 // This service is compatible with the Ethereum service interface (node.Service)
 type Indexer interface {
@@ -97,7 +93,7 @@ func (sap *Service) Sync(wg *sync.WaitGroup) error {
 		return err
 	}
 	// spin up publish worker goroutines
-	publishPayload := make(chan statediff.Payload, PayloadChanBufferSize)
+	publishPayload := make(chan statediff.Payload, eth.PayloadChanBufferSize)
 	for i := 1; i <= int(sap.Workers); i++ {
 		go sap.transform(wg, i, publishPayload)
 		log.Debugf("ethereum sync worker %d successfully spun up", i)