syz-manager, syz-hub: share repros between managers via hub

Currently hub allows managers to exchange programs from corpus.
But reproducers are not exchanged and we don't know if a crash
happens on other managers as well or not.

Allow hub to exchange reproducers.

Reproducers are stored in a separate db file with own sequence numbers.
This allows to throttle distribution of reproducers to managers,
so that they are not overloaded with reproducers and don't lose them on restarts.

Based on patch by Andrey Konovalov:
https://github.com/google/syzkaller/pull/325

Fixes #282

8 files changed, 402 insertions, 125 deletions

diff --git a/pkg/osutil/osutil.go b/pkg/osutil/osutil.go
index fecb5f95b..4fbd04506 100644
--- a/pkg/osutil/osutil.go
+++ b/pkg/osutil/osutil.go
@@ -155,3 +155,13 @@ func WriteFile(filename string, data []byte) error {
 func WriteExecFile(filename string, data []byte) error {
 	return ioutil.WriteFile(filename, data, DefaultExecPerm)
 }
+
+// Return all files in a directory.
+func ListDir(dir string) ([]string, error) {
+	f, err := os.Open(dir)
+	if err != nil {
+		return nil, err
+	}
+	defer f.Close()
+	return f.Readdirnames(-1)
+}
diff --git a/pkg/rpctype/rpctype.go b/pkg/rpctype/rpctype.go
index ab3f2243a..a198a90f4 100644
--- a/pkg/rpctype/rpctype.go
+++ b/pkg/rpctype/rpctype.go
@@ -74,18 +74,23 @@ type HubConnectArgs struct {
 
 type HubSyncArgs struct {
 	// see HubConnectArgs.
-	Client  string
-	Key     string
-	Manager string
+	Client     string
+	Key        string
+	Manager    string
+	NeedRepros bool
 	// Programs added to corpus since last sync or connect.
 	Add [][]byte
-	// Hashed of programs removed from corpus since last sync or connect.
+	// Hashes of programs removed from corpus since last sync or connect.
 	Del []string
+	// Repros found since last sync.
+	Repros [][]byte
 }
 
 type HubSyncRes struct {
 	// Set of programs from other managers.
-	Inputs [][]byte
+	Progs [][]byte
+	// Set of repros from other managers.
+	Repros [][]byte
 	// Number of remaining pending programs,
 	// if >0 manager should do sync again.
 	More int
diff --git a/syz-hub/http.go b/syz-hub/http.go
index 3ee6ff686..aa99760f3 100644
--- a/syz-hub/http.go
+++ b/syz-hub/http.go
@@ -38,17 +38,22 @@ func (hub *Hub) httpSummary(w http.ResponseWriter, r *http.Request) {
 	total := UIManager{
 		Name:   "total",
 		Corpus: len(hub.st.Corpus.Records),
+		Repros: len(hub.st.Repros.Records),
 	}
 	for name, mgr := range hub.st.Managers {
 		total.Added += mgr.Added
 		total.Deleted += mgr.Deleted
 		total.New += mgr.New
+		total.SentRepros += mgr.SentRepros
+		total.RecvRepros += mgr.RecvRepros
 		data.Managers = append(data.Managers, UIManager{
-			Name:    name,
-			Corpus:  len(mgr.Corpus.Records),
-			Added:   mgr.Added,
-			Deleted: mgr.Deleted,
-			New:     mgr.New,
+			Name:       name,
+			Corpus:     len(mgr.Corpus.Records),
+			Added:      mgr.Added,
+			Deleted:    mgr.Deleted,
+			New:        mgr.New,
+			SentRepros: mgr.SentRepros,
+			RecvRepros: mgr.RecvRepros,
 		})
 	}
 	sort.Sort(UIManagerArray(data.Managers))
@@ -70,11 +75,14 @@ type UISummaryData struct {
 }
 
 type UIManager struct {
-	Name    string
-	Corpus  int
-	Added   int
-	Deleted int
-	New     int
+	Name       string
+	Corpus     int
+	Added      int
+	Deleted    int
+	New        int
+	Repros     int
+	SentRepros int
+	RecvRepros int
 }
 
 type UIManagerArray []UIManager
@@ -102,6 +110,9 @@ var summaryTemplate = compileTemplate(`
 		<th>Added</th>
 		<th>Deleted</th>
 		<th>New</th>
+		<th>Repros</th>
+		<th>Sent</th>
+		<th>Recv</th>
 	</tr>
 	{{range $m := $.Managers}}
 	<tr>
@@ -110,6 +121,9 @@ var summaryTemplate = compileTemplate(`
 		<td>{{$m.Added}}</td>
 		<td>{{$m.Deleted}}</td>
 		<td>{{$m.New}}</td>
+		<td>{{$m.Repros}}</td>
+		<td>{{$m.SentRepros}}</td>
+		<td>{{$m.RecvRepros}}</td>
 	</tr>
 	{{end}}
 </table>
diff --git a/syz-hub/hub.go b/syz-hub/hub.go
index 6e44c51cd..d870f569b 100644
--- a/syz-hub/hub.go
+++ b/syz-hub/hub.go
@@ -90,15 +90,29 @@ func (hub *Hub) Sync(a *HubSyncArgs, r *HubSyncRes) error {
 	hub.mu.Lock()
 	defer hub.mu.Unlock()
 
-	inputs, more, err := hub.st.Sync(name, a.Add, a.Del)
+	progs, more, err := hub.st.Sync(name, a.Add, a.Del)
 	if err != nil {
 		Logf(0, "sync error: %v", err)
 		return err
 	}
-	r.Inputs = inputs
+	r.Progs = progs
 	r.More = more
-	Logf(0, "sync from %v: add=%v del=%v new=%v pending=%v",
-		name, len(a.Add), len(a.Del), len(inputs), more)
+	for _, repro := range a.Repros {
+		if err := hub.st.AddRepro(name, repro); err != nil {
+			Logf(0, "add repro error: %v", err)
+		}
+	}
+	if a.NeedRepros {
+		repro, err := hub.st.PendingRepro(name)
+		if err != nil {
+			Logf(0, "sync error: %v", err)
+		}
+		if repro != nil {
+			r.Repros = [][]byte{repro}
+		}
+	}
+	Logf(0, "sync from %v: recv: add=%v del=%v; send: progs=%v repros=%v pending=%v",
+		name, len(a.Add), len(a.Del), len(r.Progs), len(r.Repros), more)
 	return nil
 }
 
diff --git a/syz-hub/state/state.go b/syz-hub/state/state.go
index 31ee6d2ea..f91efbeb8 100644
--- a/syz-hub/state/state.go
+++ b/syz-hub/state/state.go
@@ -19,26 +19,35 @@ import (
 	"github.com/google/syzkaller/prog"
 )
 
-// State holds all internal syz-hub state including corpus and information about managers.
+// State holds all internal syz-hub state including corpus,
+// reproducers and information about managers.
 // It is persisted to and can be restored from a directory.
 type State struct {
-	seq      uint64
-	dir      string
-	Corpus   *db.DB
-	Managers map[string]*Manager
+	corpusSeq uint64
+	reproSeq  uint64
+	dir       string
+	Corpus    *db.DB
+	Repros    *db.DB
+	Managers  map[string]*Manager
 }
 
 // Manager represents one syz-manager instance.
 type Manager struct {
-	name      string
-	seq       uint64
-	dir       string
-	Connected time.Time
-	Added     int
-	Deleted   int
-	New       int
-	Calls     map[string]struct{}
-	Corpus    *db.DB
+	name          string
+	corpusSeq     uint64
+	reproSeq      uint64
+	corpusFile    string
+	corpusSeqFile string
+	reproSeqFile  string
+	ownRepros     map[string]bool
+	Connected     time.Time
+	Added         int
+	Deleted       int
+	New           int
+	SentRepros    int
+	RecvRepros    int
+	Calls         map[string]struct{}
+	Corpus        *db.DB
 }
 
 // Make creates State and initializes it from dir.
@@ -49,31 +58,8 @@ func Make(dir string) (*State, error) {
 	}
 
 	osutil.MkdirAll(st.dir)
-	var err error
-	Logf(0, "reading corpus...")
-	st.Corpus, err = db.Open(filepath.Join(st.dir, "corpus.db"))
-	if err != nil {
-		Fatalf("failed to open corpus database: %v", err)
-	}
-	Logf(0, "read %v programs", len(st.Corpus.Records))
-	for key, rec := range st.Corpus.Records {
-		if _, err := prog.CallSet(rec.Val); err != nil {
-			Logf(0, "bad file in corpus: can't parse call set: %v", err)
-			st.Corpus.Delete(key)
-			continue
-		}
-		if sig := hash.Hash(rec.Val); sig.String() != key {
-			Logf(0, "bad file in corpus: hash %v, want hash %v", key, sig.String())
-			st.Corpus.Delete(key)
-			continue
-		}
-		if st.seq < rec.Seq {
-			st.seq = rec.Seq
-		}
-	}
-	if err := st.Corpus.Flush(); err != nil {
-		Fatalf("failed to flush corpus database: %v", err)
-	}
+	st.Corpus, st.corpusSeq = loadDB(filepath.Join(st.dir, "corpus.db"), "corpus")
+	st.Repros, st.reproSeq = loadDB(filepath.Join(st.dir, "repro.db"), "repro")
 
 	managersDir := filepath.Join(st.dir, "manager")
 	osutil.MkdirAll(managersDir)
@@ -82,22 +68,10 @@ func Make(dir string) (*State, error) {
 		return nil, fmt.Errorf("failed to read %v dir: %v", managersDir, err)
 	}
 	for _, manager := range managers {
-		mgr := &Manager{
-			name: manager.Name(),
-		}
-		st.Managers[mgr.name] = mgr
-		mgr.dir = filepath.Join(managersDir, mgr.name)
-		seqStr, _ := ioutil.ReadFile(filepath.Join(mgr.dir, "seq"))
-		mgr.seq, _ = strconv.ParseUint(string(seqStr), 10, 64)
-		if st.seq < mgr.seq {
-			st.seq = mgr.seq
-		}
-		Logf(0, "reading %v corpus...", mgr.name)
-		mgr.Corpus, err = db.Open(filepath.Join(mgr.dir, "corpus.db"))
+		_, err := st.createManager(manager.Name())
 		if err != nil {
-			return nil, fmt.Errorf("failed to open manager corpus database %v: %v", mgr.dir, err)
+			return nil, err
 		}
-		Logf(0, "read %v programs", len(mgr.Corpus.Records))
 	}
 	Logf(0, "purging corpus...")
 	st.purgeCorpus()
@@ -106,29 +80,89 @@ func Make(dir string) (*State, error) {
 	return st, err
 }
 
+func loadDB(file, name string) (*db.DB, uint64) {
+	Logf(0, "reading %v...", name)
+	db, err := db.Open(file)
+	if err != nil {
+		Fatalf("failed to open %v database: %v", name, err)
+	}
+	Logf(0, "read %v programs", len(db.Records))
+	var maxSeq uint64
+	for key, rec := range db.Records {
+		if _, err := prog.CallSet(rec.Val); err != nil {
+			Logf(0, "bad file: can't parse call set: %v", err)
+			db.Delete(key)
+			continue
+		}
+		if sig := hash.Hash(rec.Val); sig.String() != key {
+			Logf(0, "bad file: hash %v, want hash %v", key, sig.String())
+			db.Delete(key)
+			continue
+		}
+		if maxSeq < rec.Seq {
+			maxSeq = rec.Seq
+		}
+	}
+	if err := db.Flush(); err != nil {
+		Fatalf("failed to flush corpus database: %v", err)
+	}
+	return db, maxSeq
+}
+
+func (st *State) createManager(name string) (*Manager, error) {
+	dir := filepath.Join(st.dir, "manager", name)
+	osutil.MkdirAll(dir)
+	mgr := &Manager{
+		name:          name,
+		corpusFile:    filepath.Join(dir, "corpus.db"),
+		corpusSeqFile: filepath.Join(dir, "seq"),
+		reproSeqFile:  filepath.Join(dir, "repro.seq"),
+		ownRepros:     make(map[string]bool),
+	}
+	mgr.corpusSeq = loadSeqFile(mgr.corpusSeqFile)
+	if st.corpusSeq < mgr.corpusSeq {
+		st.corpusSeq = mgr.corpusSeq
+	}
+	mgr.reproSeq = loadSeqFile(mgr.reproSeqFile)
+	if st.reproSeq < mgr.reproSeq {
+		st.reproSeq = mgr.reproSeq
+	}
+	var err error
+	mgr.Corpus, err = db.Open(mgr.corpusFile)
+	if err != nil {
+		return nil, fmt.Errorf("failed to open manager corpus %v: %v", mgr.corpusFile, err)
+	}
+	Logf(0, "created manager %v: corpus=%v, corpusSeq=%v, reproSeq=%v",
+		mgr.name, len(mgr.Corpus.Records), mgr.corpusSeq, mgr.reproSeq)
+	st.Managers[name] = mgr
+	return mgr, nil
+}
+
 func (st *State) Connect(name string, fresh bool, calls []string, corpus [][]byte) error {
 	mgr := st.Managers[name]
 	if mgr == nil {
-		mgr = new(Manager)
-		st.Managers[name] = mgr
-		mgr.dir = filepath.Join(st.dir, "manager", name)
-		osutil.MkdirAll(mgr.dir)
+		var err error
+		mgr, err = st.createManager(name)
+		if err != nil {
+			return err
+		}
 	}
 	mgr.Connected = time.Now()
 	if fresh {
-		mgr.seq = 0
+		mgr.corpusSeq = 0
+		mgr.reproSeq = 0
 	}
-	writeFile(filepath.Join(mgr.dir, "seq"), []byte(fmt.Sprint(mgr.seq)))
+	saveSeqFile(mgr.corpusSeqFile, mgr.corpusSeq)
+	saveSeqFile(mgr.reproSeqFile, mgr.reproSeq)
 
 	mgr.Calls = make(map[string]struct{})
 	for _, c := range calls {
 		mgr.Calls[c] = struct{}{}
 	}
 
-	corpusFile := filepath.Join(mgr.dir, "corpus.db")
-	os.Remove(corpusFile)
+	os.Remove(mgr.corpusFile)
 	var err error
-	mgr.Corpus, err = db.Open(corpusFile)
+	mgr.Corpus, err = db.Open(mgr.corpusFile)
 	if err != nil {
 		Logf(0, "failed to open corpus database: %v", err)
 		return err
@@ -153,20 +187,88 @@ func (st *State) Sync(name string, add [][]byte, del []string) ([][]byte, int, e
 		st.purgeCorpus()
 	}
 	st.addInputs(mgr, add)
-	inputs, more, err := st.pendingInputs(mgr)
+	progs, more, err := st.pendingInputs(mgr)
 	mgr.Added += len(add)
 	mgr.Deleted += len(del)
-	mgr.New += len(inputs)
-	return inputs, more, err
+	mgr.New += len(progs)
+	return progs, more, err
+}
+
+func (st *State) AddRepro(name string, repro []byte) error {
+	mgr := st.Managers[name]
+	if mgr == nil || mgr.Connected.IsZero() {
+		return fmt.Errorf("unconnected manager %v", name)
+	}
+	if _, err := prog.CallSet(repro); err != nil {
+		Logf(0, "manager %v: failed to extract call set: %v, program:\n%v",
+			mgr.name, err, string(repro))
+		return nil
+	}
+	sig := hash.String(repro)
+	if _, ok := st.Repros.Records[sig]; ok {
+		return nil
+	}
+	mgr.ownRepros[sig] = true
+	mgr.SentRepros++
+	if mgr.reproSeq == st.reproSeq {
+		mgr.reproSeq++
+		saveSeqFile(mgr.reproSeqFile, mgr.reproSeq)
+	}
+	st.reproSeq++
+	st.Repros.Save(sig, repro, st.reproSeq)
+	if err := st.Repros.Flush(); err != nil {
+		Logf(0, "failed to flush repro database: %v", err)
+	}
+	return nil
+}
+
+func (st *State) PendingRepro(name string) ([]byte, error) {
+	mgr := st.Managers[name]
+	if mgr == nil || mgr.Connected.IsZero() {
+		return nil, fmt.Errorf("unconnected manager %v", name)
+	}
+	if mgr.reproSeq == st.reproSeq {
+		return nil, nil
+	}
+	var repro []byte
+	minSeq := ^uint64(0)
+	for key, rec := range st.Repros.Records {
+		if mgr.reproSeq >= rec.Seq {
+			continue
+		}
+		if mgr.ownRepros[key] {
+			continue
+		}
+		calls, err := prog.CallSet(rec.Val)
+		if err != nil {
+			return nil, fmt.Errorf("failed to extract call set: %v\nprogram: %s", err, rec.Val)
+		}
+		if !managerSupportsAllCalls(mgr.Calls, calls) {
+			continue
+		}
+		if minSeq > rec.Seq {
+			minSeq = rec.Seq
+			repro = rec.Val
+		}
+	}
+	if repro == nil {
+		mgr.reproSeq = st.reproSeq
+		saveSeqFile(mgr.reproSeqFile, mgr.reproSeq)
+		return nil, nil
+	}
+	mgr.RecvRepros++
+	mgr.reproSeq = minSeq
+	saveSeqFile(mgr.reproSeqFile, mgr.reproSeq)
+	return repro, nil
 }
 
 func (st *State) pendingInputs(mgr *Manager) ([][]byte, int, error) {
-	if mgr.seq == st.seq {
+	if mgr.corpusSeq == st.corpusSeq {
 		return nil, 0, nil
 	}
 	var records []db.Record
 	for key, rec := range st.Corpus.Records {
-		if mgr.seq >= rec.Seq {
+		if mgr.corpusSeq >= rec.Seq {
 			continue
 		}
 		if _, ok := mgr.Corpus.Records[key]; ok {
@@ -181,7 +283,7 @@ func (st *State) pendingInputs(mgr *Manager) ([][]byte, int, error) {
 		}
 		records = append(records, rec)
 	}
-	maxSeq := st.seq
+	maxSeq := st.corpusSeq
 	more := 0
 	// Send at most that many records (rounded up to next seq number).
 	const maxRecords = 1000
@@ -196,20 +298,20 @@ func (st *State) pendingInputs(mgr *Manager) ([][]byte, int, error) {
 		more = len(records) - pos
 		records = records[:pos]
 	}
-	inputs := make([][]byte, len(records))
-	for i, rec := range records {
-		inputs[i] = rec.Val
+	progs := make([][]byte, len(records))
+	for _, rec := range records {
+		progs = append(progs, rec.Val)
 	}
-	mgr.seq = maxSeq
-	writeFile(filepath.Join(mgr.dir, "seq"), []byte(fmt.Sprint(mgr.seq)))
-	return inputs, more, nil
+	mgr.corpusSeq = maxSeq
+	saveSeqFile(mgr.corpusSeqFile, mgr.corpusSeq)
+	return progs, more, nil
 }
 
 func (st *State) addInputs(mgr *Manager, inputs [][]byte) {
 	if len(inputs) == 0 {
 		return
 	}
-	st.seq++
+	st.corpusSeq++
 	for _, input := range inputs {
 		st.addInput(mgr, input)
 	}
@@ -229,13 +331,7 @@ func (st *State) addInput(mgr *Manager, input []byte) {
 	sig := hash.String(input)
 	mgr.Corpus.Save(sig, nil, 0)
 	if _, ok := st.Corpus.Records[sig]; !ok {
-		st.Corpus.Save(sig, input, st.seq)
-	}
-}
-
-func writeFile(name string, data []byte) {
-	if err := osutil.WriteFile(name, data); err != nil {
-		Logf(0, "failed to write file %v: %v", name, err)
+		st.Corpus.Save(sig, input, st.corpusSeq)
 	}
 }
 
@@ -266,6 +362,22 @@ func managerSupportsAllCalls(mgr, prog map[string]struct{}) bool {
 	return true
 }
 
+func writeFile(name string, data []byte) {
+	if err := osutil.WriteFile(name, data); err != nil {
+		Logf(0, "failed to write file %v: %v", name, err)
+	}
+}
+
+func saveSeqFile(filename string, seq uint64) {
+	writeFile(filename, []byte(fmt.Sprint(seq)))
+}
+
+func loadSeqFile(filename string) uint64 {
+	str, _ := ioutil.ReadFile(filename)
+	seq, _ := strconv.ParseUint(string(str), 10, 64)
+	return seq
+}
+
 type recordSeqSorter []db.Record
 
 func (a recordSeqSorter) Len() int {
diff --git a/syz-hub/state/state_test.go b/syz-hub/state/state_test.go
index bfdf4b3a7..0972ad6bc 100644
--- a/syz-hub/state/state_test.go
+++ b/syz-hub/state/state_test.go
@@ -4,13 +4,16 @@
 package state
 
 import (
+	"fmt"
 	"io/ioutil"
 	"os"
+	"path/filepath"
+	"runtime"
 	"testing"
 )
 
 func TestState(t *testing.T) {
-	dir, err := ioutil.TempDir("", "syz-gce-state-test")
+	dir, err := ioutil.TempDir("", "syz-hub-state-test")
 	if err != nil {
 		t.Fatalf("failed to create temp dir: %v", err)
 	}
@@ -24,4 +27,88 @@ func TestState(t *testing.T) {
 	if err == nil {
 		t.Fatalf("synced with unconnected manager")
 	}
+	calls := []string{"read", "write"}
+	if err := st.Connect("foo", false, calls, nil); err != nil {
+		t.Fatalf("Connect failed: %v", err)
+	}
+	_, _, err = st.Sync("foo", nil, nil)
+	if err != nil {
+		t.Fatalf("Sync failed: %v", err)
+	}
+}
+
+func TestRepro(t *testing.T) {
+	dir, err := ioutil.TempDir("", "syz-hub-state-test")
+	if err != nil {
+		t.Fatalf("failed to create temp dir: %v", err)
+	}
+	defer os.RemoveAll(dir)
+
+	st, err := Make(dir)
+	if err != nil {
+		t.Fatalf("failed to make state: %v", err)
+	}
+
+	if err := st.Connect("foo", false, []string{"open", "read", "write"}, nil); err != nil {
+		t.Fatalf("Connect failed: %v", err)
+	}
+	if err := st.Connect("bar", false, []string{"open", "read", "close"}, nil); err != nil {
+		t.Fatalf("Connect failed: %v", err)
+	}
+	checkPendingRepro(t, st, "foo", "")
+	checkPendingRepro(t, st, "bar", "")
+
+	if err := st.AddRepro("foo", []byte("open()")); err != nil {
+		t.Fatalf("AddRepro failed: %v", err)
+	}
+	checkPendingRepro(t, st, "foo", "")
+	checkPendingRepro(t, st, "bar", "open()")
+	checkPendingRepro(t, st, "bar", "")
+
+	// This repro is already present.
+	if err := st.AddRepro("bar", []byte("open()")); err != nil {
+		t.Fatalf("AddRepro failed: %v", err)
+	}
+	if err := st.AddRepro("bar", []byte("read()")); err != nil {
+		t.Fatalf("AddRepro failed: %v", err)
+	}
+	if err := st.AddRepro("bar", []byte("open()\nread()")); err != nil {
+		t.Fatalf("AddRepro failed: %v", err)
+	}
+	// This does not satisfy foo's call set.
+	if err := st.AddRepro("bar", []byte("close()")); err != nil {
+		t.Fatalf("AddRepro failed: %v", err)
+	}
+	checkPendingRepro(t, st, "bar", "")
+
+	// Check how persistence works.
+	st, err = Make(dir)
+	if err != nil {
+		t.Fatalf("failed to make state: %v", err)
+	}
+	if err := st.Connect("foo", false, []string{"open", "read", "write"}, nil); err != nil {
+		t.Fatalf("Connect failed: %v", err)
+	}
+	if err := st.Connect("bar", false, []string{"open", "read", "close"}, nil); err != nil {
+		t.Fatalf("Connect failed: %v", err)
+	}
+	checkPendingRepro(t, st, "bar", "")
+	checkPendingRepro(t, st, "foo", "read()")
+	checkPendingRepro(t, st, "foo", "open()\nread()")
+	checkPendingRepro(t, st, "foo", "")
+}
+
+func checkPendingRepro(t *testing.T, st *State, name, result string) {
+	repro, err := st.PendingRepro(name)
+	if err != nil {
+		t.Fatalf("\n%v: PendingRepro failed: %v", caller(1), err)
+	}
+	if string(repro) != result {
+		t.Fatalf("\n%v: PendingRepro returned %q, want %q", caller(1), string(repro), result)
+	}
+}
+
+func caller(skip int) string {
+	_, file, line, _ := runtime.Caller(skip + 1)
+	return fmt.Sprintf("%v:%v", filepath.Base(file), line)
 }
diff --git a/syz-manager/html.go b/syz-manager/html.go
index 1d3ba0bfb..d547980a3 100644
--- a/syz-manager/html.go
+++ b/syz-manager/html.go
@@ -292,7 +292,7 @@ func (mgr *Manager) httpReport(w http.ResponseWriter, r *http.Request) {
 
 func collectCrashes(workdir string) ([]*UICrashType, error) {
 	crashdir := filepath.Join(workdir, "crashes")
-	dirs, err := readdirnames(crashdir)
+	dirs, err := osutil.ListDir(crashdir)
 	if err != nil {
 		return nil, err
 	}
@@ -329,7 +329,7 @@ func readCrash(workdir, dir string, full bool) *UICrashType {
 	modTime := stat.ModTime()
 	descFile.Close()
 
-	files, err := readdirnames(filepath.Join(crashdir, dir))
+	files, err := osutil.ListDir(filepath.Join(crashdir, dir))
 	if err != nil {
 		return nil
 	}
@@ -395,15 +395,6 @@ func readCrash(workdir, dir string, full bool) *UICrashType {
 	}
 }
 
-func readdirnames(dir string) ([]string, error) {
-	f, err := os.Open(dir)
-	if err != nil {
-		return nil, err
-	}
-	defer f.Close()
-	return f.Readdirnames(-1)
-}
-
 func trimNewLines(data []byte) []byte {
 	for len(data) > 0 && data[len(data)-1] == '\n' {
 		data = data[:len(data)-1]
diff --git a/syz-manager/manager.go b/syz-manager/manager.go
index ab6d92192..7ac547c4c 100644
--- a/syz-manager/manager.go
+++ b/syz-manager/manager.go
@@ -70,10 +70,13 @@ type Manager struct {
 	maxSignal      map[uint32]struct{}
 	corpusCover    map[uint32]struct{}
 	prios          [][]float32
+	newRepros      [][]byte
 
-	fuzzers   map[string]*Fuzzer
-	hub       *RpcClient
-	hubCorpus map[hash.Sig]bool
+	fuzzers        map[string]*Fuzzer
+	hub            *RpcClient
+	hubCorpus      map[hash.Sig]bool
+	needMoreRepros chan chan bool
+	hubReproQueue  chan *Crash
 }
 
 const (
@@ -100,6 +103,7 @@ type Crash struct {
 	desc    string
 	text    []byte
 	output  []byte
+	hub     bool // this crash was created based on a repro from hub
 }
 
 func main() {
@@ -149,6 +153,8 @@ func RunManager(cfg *mgrconfig.Config, syscalls map[int]bool) {
 		fuzzers:         make(map[string]*Fuzzer),
 		fresh:           true,
 		vmStop:          make(chan bool),
+		hubReproQueue:   make(chan *Crash), //!!! make buffered
+		needMoreRepros:  make(chan chan bool),
 	}
 
 	Logf(0, "loading corpus...")
@@ -308,6 +314,7 @@ type ReproResult struct {
 	desc0     string
 	res       *repro.Result
 	err       error
+	hub       bool // repro came from hub
 }
 
 func (mgr *Manager) vmLoop() {
@@ -373,7 +380,7 @@ func (mgr *Manager) vmLoop() {
 				Logf(1, "loop: starting repro of '%v' on instances %+v", crash.desc, vmIndexes)
 				go func() {
 					res, err := repro.Run(crash.output, mgr.cfg, mgr.vmPool, vmIndexes)
-					reproDone <- &ReproResult{vmIndexes, crash.desc, res, err}
+					reproDone <- &ReproResult{vmIndexes, crash.desc, res, err, crash.hub}
 				}()
 			}
 			for !canRepro() && len(instances) != 0 {
@@ -431,11 +438,17 @@ func (mgr *Manager) vmLoop() {
 			if res.res == nil {
 				mgr.saveFailedRepro(res.desc0)
 			} else {
-				mgr.saveRepro(res.res)
+				mgr.saveRepro(res.res, res.hub)
 			}
 		case <-shutdown:
 			Logf(1, "loop: shutting down...")
 			shutdown = nil
+		case crash := <-mgr.hubReproQueue:
+			Logf(1, "loop: get repro from hub")
+			reproQueue = append(reproQueue, crash)
+		case reply := <-mgr.needMoreRepros:
+			reply <- phase >= phaseTriagedHub &&
+				len(reproQueue)+len(pendingRepro)+len(reproducing) == 0
 		}
 	}
 }
@@ -490,7 +503,7 @@ func (mgr *Manager) runInstance(index int) (*Crash, error) {
 		// syz-fuzzer exited, but it should not.
 		desc = "lost connection to test machine"
 	}
-	return &Crash{index, desc, text, output}, nil
+	return &Crash{index, desc, text, output, false}, nil
 }
 
 func (mgr *Manager) isSuppressed(crash *Crash) bool {
@@ -617,11 +630,14 @@ func (mgr *Manager) saveFailedRepro(desc string) {
 	}
 }
 
-func (mgr *Manager) saveRepro(res *repro.Result) {
+func (mgr *Manager) saveRepro(res *repro.Result, hub bool) {
 	res.Report = mgr.symbolizeReport(res.Report)
 	dir := filepath.Join(mgr.crashdir, hash.String([]byte(res.Desc)))
 	osutil.MkdirAll(dir)
 
+	if err := osutil.WriteFile(filepath.Join(dir, "description"), []byte(res.Desc+"\n")); err != nil {
+		Logf(0, "failed to write crash: %v", err)
+	}
 	opts := fmt.Sprintf("# %+v\n", res.Opts)
 	prog := res.Prog.Serialize()
 	osutil.WriteFile(filepath.Join(dir, "repro.prog"), append([]byte(opts), prog...))
@@ -653,6 +669,13 @@ func (mgr *Manager) saveRepro(res *repro.Result) {
 		}
 	}
 
+	// Append this repro to repro list to send to hub if it didn't come from hub originally.
+	if !hub {
+		mgr.mu.Lock()
+		mgr.newRepros = append(mgr.newRepros, prog)
+		mgr.mu.Unlock()
+	}
+
 	if mgr.dash != nil {
 		var maintainers []string
 		guiltyFile := report.ExtractGuiltyFile(res.Report)
@@ -982,7 +1005,14 @@ func (mgr *Manager) hubSync() {
 		a.Del = append(a.Del, sig.String())
 	}
 	for {
+		a.Repros = mgr.newRepros
+
 		mgr.mu.Unlock()
+
+		needReproReply := make(chan bool)
+		mgr.needMoreRepros <- needReproReply
+		a.NeedRepros = <-needReproReply
+
 		r := new(HubSyncRes)
 		if err := mgr.hub.Call("Hub.Sync", a, r); err != nil {
 			mgr.mu.Lock()
@@ -991,9 +1021,21 @@ func (mgr *Manager) hubSync() {
 			mgr.hub = nil
 			return
 		}
+
+		reproDropped := 0
+		for _, repro := range r.Repros {
+			_, err := prog.Deserialize(repro)
+			if err != nil {
+				reproDropped++
+				continue
+			}
+			mgr.hubReproQueue <- &Crash{-1, "external repro", nil, repro, true}
+		}
+
 		mgr.mu.Lock()
+		mgr.newRepros = nil
 		dropped := 0
-		for _, inp := range r.Inputs {
+		for _, inp := range r.Progs {
 			_, err := prog.Deserialize(inp)
 			if err != nil {
 				dropped++
@@ -1007,10 +1049,12 @@ func (mgr *Manager) hubSync() {
 		mgr.stats["hub add"] += uint64(len(a.Add))
 		mgr.stats["hub del"] += uint64(len(a.Del))
 		mgr.stats["hub drop"] += uint64(dropped)
-		mgr.stats["hub new"] += uint64(len(r.Inputs) - dropped)
-		Logf(0, "hub sync: add %v, del %v, drop %v, new %v, more %v",
-			len(a.Add), len(a.Del), dropped, len(r.Inputs)-dropped, r.More)
-		if len(r.Inputs)+r.More == 0 {
+		mgr.stats["hub new"] += uint64(len(r.Progs) - dropped)
+		mgr.stats["hub sent repros"] += uint64(len(a.Repros))
+		mgr.stats["hub recv repros"] += uint64(len(r.Repros) - reproDropped)
+		Logf(0, "hub sync: send: add %v, del %v, repros: %v; recv: progs: drop %v, new %v, repros: drop: %v, new %v; more %v",
+			len(a.Add), len(a.Del), len(a.Repros), dropped, len(r.Progs)-dropped, reproDropped, len(r.Repros)-reproDropped, r.More)
+		if len(r.Progs)+r.More == 0 {
 			break
 		}
 		a.Add = nil