syz-cluster/email-reporter: poll an LKML archive for replies

Add the functionality to poll the specified LKML git archive for the new
user replies under the bot's messages.

Check out the repository and pull it once in the specified time period.
Keep track of the latest processed commit hash.

Track the original report ID via In-Reply-To references.

Author: a-nogikh

syz-cluster/email-reporter/deployment.yaml

@@ -22,6 +22,8 @@ spec:
         volumeMounts:
           - name: config-volume
             mountPath: /config
+          - name: reporter-lore-disk
+            mountPath: /lore-repo
         resources:
           requests:
             cpu: 2
@@ -33,3 +35,6 @@ spec:
       - name: config-volume
         configMap:
           name: global-config
+      - name: reporter-lore-disk
+        persistentVolumeClaim:
+          claimName: reporter-lore-disk-claim

syz-cluster/email-reporter/handler_test.go

@@ -18,7 +18,7 @@ import (
 func TestModerationReportFlow(t *testing.T) {
 	env, ctx := app.TestEnvironment(t)
 	testSeries := controller.DummySeries()
-	handler, emailServer := setupHandlerTest(t, env, ctx, testSeries)
+	handler, _, emailServer := setupHandlerTest(t, env, ctx, testSeries)
 
 	report, err := handler.PollAndReport(ctx)
 	assert.NoError(t, err)
@@ -61,7 +61,7 @@ func TestModerationReportFlow(t *testing.T) {
 func TestInvalidReply(t *testing.T) {
 	env, ctx := app.TestEnvironment(t)
 	testSeries := controller.DummySeries()
-	handler, emailServer := setupHandlerTest(t, env, ctx, testSeries)
+	handler, _, emailServer := setupHandlerTest(t, env, ctx, testSeries)
 
 	report, err := handler.PollAndReport(ctx)
 	assert.NoError(t, err)
@@ -117,7 +117,7 @@ Unknown command
 }
 
 func setupHandlerTest(t *testing.T, env *app.AppEnvironment, ctx context.Context,
-	series *api.Series) (*Handler, *fakeSender) {
+	series *api.Series) (*Handler, *api.ReporterClient, *fakeSender) {
 	client := controller.TestServer(t, env)
 	controller.FakeSeriesWithFindings(t, ctx, env, client, series)
 
@@ -126,13 +126,14 @@ func setupHandlerTest(t *testing.T, env *app.AppEnvironment, ctx context.Context
 	assert.NoError(t, err)
 
 	emailServer := makeFakeSender()
+	reporterClient := reporter.TestServer(t, env)
 	handler := &Handler{
 		reporter:    api.LKMLReporter,
-		apiClient:   reporter.TestServer(t, env),
+		apiClient:   reporterClient,
 		emailConfig: testEmailConfig,
 		sender:      emailServer.send,
 	}
-	return handler, emailServer
+	return handler, reporterClient, emailServer
 }
 
 type fakeSender struct {

syz-cluster/email-reporter/kustomization.yaml

@@ -3,3 +3,4 @@
 
 resources:
 - deployment.yaml
+- lore-disk-pvc.yaml

syz-cluster/email-reporter/lore-disk-pvc.yaml

@@ -0,0 +1,14 @@
+# Copyright 2025 syzkaller project authors. All rights reserved.
+# Use of this source code is governed by Apache 2 LICENSE that can be found in the LICENSE file.
+
+apiVersion: v1
+kind: PersistentVolumeClaim
+metadata:
+  name: reporter-lore-disk-claim
+spec:
+  accessModes:
+    - ReadWriteOnce
+  resources:
+    requests:
+      storage: 16Gi
+  storageClassName: standard

syz-cluster/email-reporter/main.go

@@ -13,14 +13,19 @@ import (
 	"github.com/google/syzkaller/pkg/email"
 	"github.com/google/syzkaller/syz-cluster/pkg/api"
 	"github.com/google/syzkaller/syz-cluster/pkg/app"
+	"golang.org/x/sync/errgroup"
 )
 
 // TODO: add extra sanity checks that would prevent flooding the mailing lists:
 // - this pod may crash and be restarted by K8S: this complicates accounting,
 // - the send operation might return an error, yet an email would be actually sent: back off on errors?
 
-// How often to check whether there are new emails to be sent.
-const pollPeriod = 30 * time.Second
+const (
+	// How often to check whether there are new emails to be sent.
+	senderPollPeriod = 30 * time.Second
+	// How often to check whether there are new incoming emails.
+	fetcherPollPeriod = 2 * time.Minute
+)
 
 func main() {
 	ctx := context.Background()
@@ -42,20 +47,34 @@ func main() {
 		emailConfig: cfg.EmailReporting,
 		sender:      sender.Send,
 	}
-	emailStream := NewLoreEmailStreamer()
-	ch := make(chan *email.Email, 16)
-	go func() {
-		for newEmail := range ch {
+	msgCh := make(chan *email.Email, 16)
+	eg, loopCtx := errgroup.WithContext(ctx)
+	if cfg.EmailReporting.LoreArchiveURL != "" {
+		fetcher := NewLKMLEmailStream("/lore-repo",
+			cfg.EmailReporting.LoreArchiveURL, reporterClient, msgCh)
+		eg.Go(func() error { return fetcher.Loop(loopCtx, fetcherPollPeriod) })
+	}
+	eg.Go(func() error {
+		for {
+			var newEmail *email.Email
+			select {
+			case newEmail = <-msgCh:
+			case <-loopCtx.Done():
+				return nil
+			}
 			log.Printf("received email %q", newEmail.MessageID)
-			err := handler.IncomingEmail(ctx, newEmail)
+			err := handler.IncomingEmail(loopCtx, newEmail)
 			if err != nil {
 				// Note that we just print an error and go on instead of retrying.
 				// Some retrying may be reasonable, but it also comes with a risk of flooding
 				// the mailing lists.
 				app.Errorf("email %q: failed to process: %v", newEmail.MessageID, err)
 			}
 		}
-	}()
-	go handler.PollReportsLoop(ctx, pollPeriod)
-	go emailStream.Loop(ctx, ch)
+	})
+	eg.Go(func() error {
+		handler.PollReportsLoop(loopCtx, senderPollPeriod)
+		return nil
+	})
+	eg.Wait()
 }

syz-cluster/email-reporter/stream.go

@@ -5,28 +5,123 @@ package main
 
 import (
 	"context"
+	"fmt"
+	"log"
+	"time"
 
 	"github.com/google/syzkaller/pkg/email"
+	"github.com/google/syzkaller/pkg/email/lore"
+	"github.com/google/syzkaller/pkg/vcs"
+	"github.com/google/syzkaller/syz-cluster/pkg/api"
+	"github.com/google/syzkaller/syz-cluster/pkg/app"
 )
 
-// TODO: there's a lot of common code with series-tracker.
-// TODO: alternatively, we could parse the whole archive and track each email via In-Reply-To to the original email.
+type LKMLEmailStream struct {
+	reporterName   string
+	repoURL        string
+	repoFolder     string
+	client         *api.ReporterClient
+	newMessages    chan *email.Email
+	lastCommitDate time.Time
+	lastCommit     string
+}
 
-type LoreEmailStreamer struct {
+func NewLKMLEmailStream(repoFolder, repoURL string, client *api.ReporterClient,
+	writeTo chan *email.Email) *LKMLEmailStream {
+	return &LKMLEmailStream{
+		reporterName: api.LKMLReporter,
+		repoURL:      repoURL,
+		repoFolder:   repoFolder,
+		client:       client,
+		newMessages:  writeTo,
+	}
 }
 
-func NewLoreEmailStreamer() *LoreEmailStreamer {
-	return &LoreEmailStreamer{}
+const (
+	// Don't consider older replies.
+	relevantPeriod = 7 * 24 * time.Hour
+)
+
+func (s *LKMLEmailStream) Loop(ctx context.Context, pollPeriod time.Duration) error {
+	last, err := s.client.LastReply(ctx, s.reporterName)
+	if err != nil {
+		return fmt.Errorf("failed to query the last reply: %w", err)
+	}
+	// We assume that the archive mostly consists of relevant emails, so after the restart
+	// we just start with the last saved message's date.
+	s.lastCommitDate = time.Now().Add(-relevantPeriod)
+	if last != nil && last.Time.After(s.lastCommitDate) {
+		s.lastCommitDate = last.Time
+	}
+	for {
+		err := s.fetchMessages(ctx)
+		if err != nil {
+			// Occasional errors are fine.
+			log.Printf("failed to poll the lore archive messages: %v", err)
+		}
+		select {
+		case <-ctx.Done():
+			return nil
+		case <-time.After(pollPeriod):
+		}
+	}
 }
 
-func (s *LoreEmailStreamer) Loop(ctx context.Context, writeTo chan *email.Email) {
-	<-ctx.Done()
+func (s *LKMLEmailStream) fetchMessages(ctx context.Context) error {
+	gitRepo := vcs.NewLKMLRepo(s.repoFolder)
+	_, err := gitRepo.Poll(s.repoURL, "master")
+	if err != nil {
+		return err
+	}
+	var messages []lore.EmailReader
+	if s.lastCommit != "" {
+		// If it's not the first iteration, it's better to rely on the last commit hash.
+		messages, err = lore.ReadArchive(gitRepo, s.lastCommit, time.Time{})
+	} else {
+		messages, err = lore.ReadArchive(gitRepo, "", s.lastCommitDate)
+	}
+	if err != nil {
+		return err
+	}
+	// From oldest to newest.
+	for i := len(messages) - 1; i >= 0; i-- {
+		msg := messages[i]
+		parsed, err := msg.Parse(nil, nil)
+		if err != nil || parsed == nil {
+			log.Printf("failed to parse the email from hash %q: %v", msg.Hash, err)
+			continue
+		}
+		if msg.CommitDate.After(s.lastCommitDate) {
+			s.lastCommitDate = msg.CommitDate
+		}
+		s.lastCommit = msg.Hash
 
-	// !! We assume that the archive mostly consists of relevant emails.
-	// 1. Query the last recorded discussion via API.
-	// 2. Poll the lore archive and query the emails starting from the date returned in (1).
-	// 3. Parse the email using email.Parse().
-	// 4. Report the new email via API, figure out which report was involved, save report ID to msg's BugIDs.
-	// 5. Push to the channel only if the message has not been seen before.
-	//    Also, we probablty don't want to react to old messages (e.g. > 1 day from now).
+		// We cannot fully trust the date specified in the message itself, so let's sanitize it
+		// using the commit date. It will at least help us prevent weird client.lastReply() responses.
+		messageDate := parsed.Date
+		if messageDate.After(msg.CommitDate) {
+			messageDate = msg.CommitDate
+		}
+		resp, err := s.client.RecordReply(ctx, &api.RecordReplyReq{
+			MessageID: parsed.MessageID,
+			InReplyTo: parsed.InReplyTo,
+			Reporter:  s.reporterName,
+			Time:      messageDate,
+		})
+		if err != nil || resp == nil {
+			// TODO: retry?
+			app.Errorf("failed to report email %q: %v", parsed.MessageID, err)
+			continue
+		} else if resp.ReportID != "" {
+			if !resp.New {
+				continue
+			}
+			parsed.BugIDs = []string{resp.ReportID}
+		}
+		select {
+		case s.newMessages <- parsed:
+		case <-ctx.Done():
+		}
+	}
+	return nil
 }