1 files changed, 21 insertions, 12 deletions
diff --git a/tools/syz-lore/query_lkml.go b/tools/syz-lore/query_lkml.go
index d6e20bf5a..ff1dbe200 100644
--- a/tools/syz-lore/query_lkml.go
+++ b/tools/syz-lore/query_lkml.go
@@ -4,7 +4,6 @@
 package main
 
 import (
-	"bytes"
 	"context"
 	"encoding/json"
 	"flag"
@@ -14,6 +13,8 @@ import (
 	"runtime"
 	"strings"
 	"sync"
+	"sync/atomic"
+	"time"
 
 	"github.com/google/syzkaller/dashboard/dashapi"
 	"github.com/google/syzkaller/pkg/email"
@@ -21,6 +22,7 @@ import (
 	"github.com/google/syzkaller/pkg/hash"
 	"github.com/google/syzkaller/pkg/osutil"
 	"github.com/google/syzkaller/pkg/tool"
+	"github.com/google/syzkaller/pkg/vcs"
 	"golang.org/x/sync/errgroup"
 )
 
@@ -106,7 +108,7 @@ func processArchives(dir string, emails, domains []string) []*lore.Thread {
 		tool.Failf("failed to read directory: %v", err)
 	}
 	threads := runtime.NumCPU()
-	messages := make(chan *lore.EmailReader, threads*2)
+	messages := make(chan lore.EmailReader, threads*2)
 	wg := sync.WaitGroup{}
 	g, _ := errgroup.WithContext(context.Background())
 
@@ -120,28 +122,32 @@ func processArchives(dir string, emails, domains []string) []*lore.Thread {
 		wg.Add(1)
 		g.Go(func() error {
 			defer wg.Done()
-			return lore.ReadArchive(path, messages)
+			repo := vcs.NewLKMLRepo(path)
+			list, err := lore.ReadArchive(repo, time.Time{})
+			if err != nil {
+				return err
+			}
+			for _, reader := range list {
+				messages <- reader
+			}
+			return nil
 		})
 	}
 
 	// Set up some worker threads.
 	var repoEmails []*email.Email
 	var mu sync.Mutex
+	var skipped atomic.Int64
 	for i := 0; i < threads; i++ {
 		g.Go(func() error {
 			for rawMsg := range messages {
-				body, err := rawMsg.Extract()
-				if err != nil {
-					continue
-				}
-				msg, err := email.Parse(bytes.NewReader(body), emails, nil, domains)
+				msg, err := rawMsg.Parse(emails, domains)
 				if err != nil {
+					// There are many broken messages in LKML,
+					// no sense to print them all each time.
+					skipped.Add(1)
 					continue
 				}
-				// Keep memory consumption low.
-				msg.Body = ""
-				msg.Patch = ""
-
 				mu.Lock()
 				repoEmails = append(repoEmails, msg)
 				mu.Unlock()
@@ -156,6 +162,9 @@ func processArchives(dir string, emails, domains []string) []*lore.Thread {
 	if err := g.Wait(); err != nil {
 		tool.Failf("%s", err)
 	}
+	if cnt := skipped.Load(); cnt > 0 {
+		log.Printf("skipped %d messages because of parsing errors", cnt)
+	}
 
 	list := lore.Threads(repoEmails)
 	log.Printf("collected %d email threads", len(list))