service.go

@@ -820,12 +820,17 @@ func (sds *Service) writeStateDiff(block *types.Block, parentRoot common.Hash, p
 		return err
 	}
 
+	var nodeMtx, ipldMtx sync.Mutex
 	nodeSink := func(node types2.StateLeafNode) error {
 		defer metrics.UpdateDuration(time.Now(), metrics.IndexerMetrics.OutputTimer)
+		nodeMtx.Lock()
+		defer nodeMtx.Unlock()
 		return sds.indexer.PushStateNode(tx, node, block.Hash().String())
 	}
 	ipldSink := func(c types2.IPLD) error {
 		defer metrics.UpdateDuration(time.Now(), metrics.IndexerMetrics.IPLDOutputTimer)
+		ipldMtx.Lock()
+		defer ipldMtx.Unlock()
 		return sds.indexer.PushIPLD(tx, c)
 	}