func (s *session) exclude(blamed []int) error {
	if len(blamed) == 0 {
		return errors.New("exclude called with no blamed peers")
	}
	blamed = uniq(blamed)
	defer s.mu.Unlock()
	s.mu.Lock()
	close(s.runs[s.run].rerunning)
	s.run++
	log.Printf("excluding %v", blamed)
	for _, pid := range blamed {
		log.Printf("excluding %v\n", s.clients[pid].raddr())
		close(s.clients[pid].blamed)
		s.excluded = append(s.excluded, s.clients[pid])
		s.clients[pid].cancel()
		s.clients[pid].conn.Close()
		s.mtot -= s.clients[pid].pr.MessageCount
		s.clients[pid] = nil
	}
	clients := s.clients[:0]
	for _, c := range s.clients {
		if c != nil {
			clients = append(clients, c)
		}
	}
	sort.Slice(clients, func(i, j int) bool {
		id1 := clients[i].pr.Identity[:]
		id2 := clients[j].pr.Identity[:]
		return bytes.Compare(id1, id2) < 0
	})
	s.clients = clients
	if len(s.clients) < s.minp {
		return fmt.Errorf("too few peers (%v < %v) to continue session",
			len(s.clients), s.minp)
	}
	s.vk = s.vk[:len(s.clients)]
	s.mcounts = s.mcounts[:len(s.clients)]
	s.pids = make(map[string]int)
	for i, c := range s.clients {
		mix, err := s.newm()
		if err != nil {
			return err
		}
		s.vk[i] = c.pr.Identity
		s.mcounts[i] = c.pr.MessageCount
		s.pids[string(c.pr.Identity)] = i
		c.ke = nil
		c.sr = nil
		c.dc = nil
		c.mix = mix
	}
	mix, err := s.newm()
	if err != nil {
		return err
	}
	s.runs = append(s.runs, runState{
		allKEs:    make(chan struct{}),
		allSRs:    make(chan struct{}),
		allDCs:    make(chan struct{}),
		allConfs:  make(chan struct{}),
		allRSs:    make(chan struct{}),
		blaming:   make(chan struct{}),
		rerunning: make(chan struct{}),
	})
	s.roots = nil
	s.msgses = messages.NewSession(s.sid, s.run, s.vk)
	s.br = messages.BeginRun(s.vk, s.mcounts, s.sid)
	s.mix = mix
	for _, c := range s.clients {
		select {
		case c.out <- s.br:
		case <-c.done:
		}
	}
	return nil
}
exclude removes blamed peers from the session so the next run can proceed.
BRs are written to each remaining client's out channel to be sent by the handler.
package server
import (
	"bytes"
	"compress/flate"
	"context"
	"crypto/rand"
	"encoding"
	"encoding/gob"
	"encoding/json"
	"errors"
	"fmt"
	"log"
	"math/big"
	"net"
	"path/filepath"
	"runtime"
	"runtime/debug"
	"runtime/trace"
	"sort"
	"sync"
	"time"
	"decred.org/cspp/chacha20prng"
	"decred.org/cspp/dcnet"
	"decred.org/cspp/messages"
	"decred.org/cspp/solver"
	"decred.org/cspp/x25519"
	"golang.org/x/crypto/ed25519"
	"golang.org/x/sync/errgroup"
)
const (
	pairTimeout = 24 * time.Hour
	sendTimeout = time.Second
	recvTimeout = 5 * time.Second
)
type Mixer interface {
	encoding.BinaryMarshaler
	encoding.BinaryUnmarshaler
	Mix(m []byte)
	Confirm(confirm interface{}, pid int) error
}
type Joiner interface {
	Join(unmixed []byte, pid int) error
	ValidateUnmixed(unmixed []byte, mcount int) error
}
type Shuffler interface {
	Shuffle()
}
type PublishMixer interface {
	PublishMix(ctx context.Context) error
}
type Server struct {
	svrState   uint32  
	svrStateMu sync.Mutex
	sidPRNG    *chacha20prng.Reader  
	msize      int
	newm       NewMixer
	epoch      time.Duration
	minPeers   int
	pairings   map[string][]*client
	pairingsMu sync.Mutex
	report *json.Encoder
}
type runState struct {
	keCount   uint32
	srCount   uint32
	dcCount   uint32
	confCount uint32
	rsCount   uint32
	allKEs   chan struct{}
	allSRs   chan struct{}
	allDCs   chan struct{}
	allConfs chan struct{}
	allRSs   chan struct{}
	blaming   chan struct{}
	rerunning chan struct{}
}
type session struct {
	runs []runState
	sid    []byte
	msgses *messages.Session
	br     *messages.BR
	msize  int
	minp   int
	newm   func() (Mixer, error)
	mix    Mixer
	run      int
	mtot     int
	clients  []*client
	excluded []*client
	vk       []ed25519.PublicKey
	mcounts  []int
	roots    []*big.Int
	pids map[string]int
	mu   sync.Mutex
	report *json.Encoder
}
type client struct {
	conn   net.Conn
	zw     *flate.Writer
	dec    *gob.Decoder  
	enc    *gob.Encoder  
	sesc   chan *session
	pr     *messages.PR
	ke     *messages.KE
	sr     *messages.SR
	dc     *messages.DC
	cm     *messages.CM
	rs     *messages.RS
	mix    Mixer
	out    chan interface{}
	blamed chan struct{}
	done   chan struct{}
	cancel func()
}
type NewMixer func(desc []byte) (Mixer, error)
func New(msize int, newm NewMixer, epoch time.Duration) (*Server, error) {
	seed := make([]byte, 32)
	_, err := rand.Read(seed)
	if err != nil {
		return nil, err
	}
	s := &Server{
		sidPRNG:  chacha20prng.New(seed, 0),
		msize:    msize,
		newm:     newm,
		epoch:    epoch,
		minPeers: 3,
		pairings: make(map[string][]*client),
	}
	return s, nil
}
func (s *Server) SetReportEncoder(enc *json.Encoder) {
	s.report = enc
}
func (s *Server) SetMinPeers(min int) {
	s.minPeers = min
}
func (s *Server) Run(ctx context.Context, lis net.Listener) error {
	s.svrStateMu.Lock()
	if s.svrState == 2 {
		s.svrStateMu.Unlock()
		return errors.New("server cannot be reused")
	}
	g, ctx := errgroup.WithContext(ctx)
	g.Go(func() error {
		if s.svrState == 0 {
			s.svrState = 1
			s.svrStateMu.Unlock()
			return s.pairSessions(ctx)
		}
		s.svrStateMu.Unlock()
		return nil
	})
	g.Go(func() error {
		var wg sync.WaitGroup
		for {
			conn, err := lis.Accept()
			if err != nil {
				wg.Wait()
				return err
			}
			log.Printf("serving %s", conn.RemoteAddr())
			ctx, task := trace.NewTask(ctx, "serveConn")
			wg.Add(1)
			go func() {
				var err error
				defer func() {
					if r := recover(); r != nil {
						log.Printf("recovered: %v", r)
						debug.PrintStack()
					}
					conn.Close()
					if err == nil {
						log.Printf("closed %v", conn.RemoteAddr())
					} else {
						log.Printf("closed %v: %v", conn.RemoteAddr(), err)
					}
					task.End()
					wg.Done()
				}()
				trace.Logf(ctx, "", "serving %v", conn.RemoteAddr())
				err = s.serveConn(ctx, conn)
				trace.Logf(ctx, "", "closed %v: %v", conn.RemoteAddr(), err)
			}()
		}
	})
	return g.Wait()
}
func (s *Server) serveConn(ctx context.Context, conn net.Conn) error {
	zr := flate.NewReader(conn)
	zw, _ := flate.NewWriter(conn, flate.DefaultCompression)
	dec := gob.NewDecoder(zr)
	enc := gob.NewEncoder(zw)
	pr := new(messages.PR)
	if err := conn.SetReadDeadline(time.Now().Add(recvTimeout)); err != nil {
		return err
	}
	err := dec.Decode(pr)
	if err != nil {
		return fmt.Errorf("read PR: %v", err)
	}
	if len(pr.Identity) != ed25519.PublicKeySize || !pr.VerifySignature(pr.Identity) {
		return errors.New("invalid identity")
	}
	if pr.MessageCount < 0 {
		return errors.New("negative message count")
	}
	log.Printf("recv(%v) PR Identity:%x PairCommitment:%x MessageCount:%d Unmixed:%x",
		conn.RemoteAddr(), pr.Identity, pr.PairCommitment, pr.MessageCount, pr.Unmixed)
	mix, err := s.newm(pr.PairCommitment)
	if err != nil {
		return fmt.Errorf("unable to begin mix: %v", err)
	}
	ctx, cancel := context.WithCancel(ctx)
	defer cancel()
	c := &client{
		conn:   conn,
		zw:     zw,
		dec:    dec,
		enc:    enc,
		sesc:   make(chan *session, 1),
		pr:     pr,
		mix:    mix,
		out:    make(chan interface{}, 1),
		blamed: make(chan struct{}),
		done:   make(chan struct{}),
		cancel: cancel,
	}
	go func() {
		<-ctx.Done()
		close(c.done)
	}()
	if j, ok := mix.(Joiner); ok {
		err = j.ValidateUnmixed(pr.Unmixed, pr.MessageCount)
		if err != nil {
			c.sendDeadline(invalidUnmixed, sendTimeout)
			return err
		}
	} else if pr.Unmixed != nil {
		return errors.New("cannot join unmixed data")
	}
	var PRRemoved bool
	defer func() {
		if !PRRemoved {
			s.removePR(pr.PairCommitment, c)
		}
	}()
	s.pairingsMu.Lock()
	s.pairings[string(pr.PairCommitment)] = append(s.pairings[string(pr.PairCommitment)], c)
	s.pairingsMu.Unlock()
	if err := conn.SetReadDeadline(time.Now().Add(pairTimeout)); err != nil {
		return err
	}
	var ses *session
	ke := new(messages.KE)
	readErr := make(chan error, 1)
	go c.read(ke, readErr)
	select {
	case <-ctx.Done():
		return ctx.Err()
	case err := <-readErr:
		if err == nil {
			return fmt.Errorf("%v: read message before run started", c.conn.RemoteAddr())
		}
		return err
	case ses = <-c.sesc:
		err := c.sendDeadline(ses.br, sendTimeout)
		if err != nil {
			return err
		}
	}
	select {
	case <-ctx.Done():
		return ctx.Err()
	case err := <-readErr:
		s.removePR(pr.PairCommitment, c)
		PRRemoved = true
		if err != nil {
			return err
		}
	}
	for run := 0; ; run++ {
		err := func() error {
			defer trace.StartRegion(ctx, "run").End()
			trace.Logf(ctx, "", "sid=%x run=%d client=%v", ses.sid, run, c.conn.RemoteAddr())
			return c.run(ctx, run, ses, ke)
		}()
		ke = nil
		if err == nil {
			return nil
		}
		if errors.Is(err, errRerun) {
			log.Printf("client %v: rerunning session %x", c.conn.RemoteAddr(), ses.sid)
			continue
		}
		return err
	}
}
func (s *Server) removePR(commitment []byte, c *client) {
	defer s.pairingsMu.Unlock()
	s.pairingsMu.Lock()
	clients := s.pairings[string(commitment)]
	for i := range clients {
		if clients[i] == c {
			clients[i] = clients[len(clients)-1]
			s.pairings[string(commitment)] = clients[:len(clients)-1]
			c.cancel()
			log.Printf("removed %v from pairing queue", c.conn.RemoteAddr())
			break
		}
	}
}
func (s *Server) pairSessions(ctx context.Context) error {
	var wg sync.WaitGroup
	defer wg.Wait()
	ticker := time.NewTicker(s.epoch)
	defer ticker.Stop()
	for {
		select {
		case <-ctx.Done():
			return ctx.Err()
		case <-ticker.C:
			log.Print("epoch tick")
		}
		var pairs []*session
		s.pairingsMu.Lock()
		for commitment, clients := range s.pairings {
			if len(clients) < s.minPeers {
				continue
			}
			clients = append(clients[:0:0], clients...)
			pairCommitment := []byte(commitment)
			newm := func() (Mixer, error) {
				return s.newm(pairCommitment)
			}
			mix, err := newm()
			if err != nil {
				log.Printf("failed to create mix object: %v", err)
				continue
			}
			delete(s.pairings, commitment)
			sort.Slice(clients, func(i, j int) bool {
				id1 := clients[i].pr.Identity[:]
				id2 := clients[j].pr.Identity[:]
				return bytes.Compare(id1, id2) < 0
			})
			vk := make([]ed25519.PublicKey, len(clients))
			mcounts := make([]int, len(clients))
			pids := make(map[string]int)
			totalMessages := 0
			for i := range clients {
				vk[i] = clients[i].pr.Identity
				pids[string(vk[i])] = i
				mcounts[i] = clients[i].pr.MessageCount
				totalMessages += clients[i].pr.MessageCount
			}
			sid := s.sidPRNG.Next(32)
			ses := &session{
				sid:     sid,
				msgses:  messages.NewSession(sid, 0, vk),
				br:      messages.BeginRun(vk, mcounts, sid),
				msize:   s.msize,
				minp:    s.minPeers,
				newm:    newm,
				mix:     mix,
				mtot:    totalMessages,
				clients: clients,
				vk:      vk,
				mcounts: mcounts,
				pids:    pids,
				report:  s.report,
			}
			ses.runs = append(ses.runs, runState{
				allKEs:    make(chan struct{}),
				allSRs:    make(chan struct{}),
				allDCs:    make(chan struct{}),
				allConfs:  make(chan struct{}),
				allRSs:    make(chan struct{}),
				blaming:   make(chan struct{}),
				rerunning: make(chan struct{}),
			})
			pairs = append(pairs, ses)
		}
		s.pairingsMu.Unlock()
		for _, ses := range pairs {
			ses.start(ctx, &wg)
		}
	}
}
func (s *session) start(ctx context.Context, wg *sync.WaitGroup) {
	for _, c := range s.clients {
		log.Printf("peer %x paired with session %x", c.pr.Identity, s.sid)
		c.sesc <- s
	}
	run := func(i int) error {
		defer trace.StartRegion(ctx, "run").End()
		err := s.doRun(ctx)
		log.Printf("session %x run %d ended: %v", s.sid, i, err)
		return err
	}
	wg.Add(1)
	defer func() {
		if r := recover(); r != nil {
			log.Printf("recovered: %v", r)
			debug.PrintStack()
		}
		for _, c := range s.clients {
			c.cancel()
		}
		wg.Done()
	}()
	for i := 0; ; i++ {
		err := run(i)
		var b blamer
		if errors.As(err, &b) {
			err = s.exclude(b.Blame())
			if err != nil {
				log.Printf("cannot continue after exclusion: %v", err)
				return
			}
			continue
		}
		if err != nil {
			s.abortSession(err)
		}
		return
	}
}
func uniq(blamed []int) []int {
	sort.Ints(blamed)
	n := len(blamed)
	blamed = blamed[:1]
	last := blamed[0]
	for _, pid := range blamed[1:n] {
		if pid != last {
			blamed = append(blamed, pid)
			last = pid
		}
	}
	return blamed
}