syncthing/protocol/protocol.go

// Copyright (C) 2014 Jakob Borg and other contributors. All rights reserved.
// Use of this source code is governed by an MIT-style license that can be
// found in the LICENSE file.

package protocol

import (
	"bufio"
	"compress/flate"
	"errors"
	"fmt"
	"io"
	"sync"
	"time"

	"github.com/calmh/syncthing/xdr"
)

const BlockSize = 128 * 1024

const (
	messageTypeClusterConfig = 0
	messageTypeIndex         = 1
	messageTypeRequest       = 2
	messageTypeResponse      = 3
	messageTypePing          = 4
	messageTypePong          = 5
	messageTypeIndexUpdate   = 6
)

const (
	stateInitial = iota
	stateCCRcvd
	stateIdxRcvd
)

const (
	FlagDeleted    uint32 = 1 << 12
	FlagInvalid           = 1 << 13
	FlagDirectory         = 1 << 14
	FlagNoPermBits        = 1 << 15
)

const (
	FlagShareTrusted  uint32 = 1 << 0
	FlagShareReadOnly        = 1 << 1
	FlagShareBits            = 0x000000ff
)

var (
	ErrClusterHash = fmt.Errorf("configuration error: mismatched cluster hash")
	ErrClosed      = errors.New("connection closed")
)

type Model interface {
	// An index was received from the peer node
	Index(nodeID string, repo string, files []FileInfo)
	// An index update was received from the peer node
	IndexUpdate(nodeID string, repo string, files []FileInfo)
	// A request was made by the peer node
	Request(nodeID string, repo string, name string, offset int64, size int) ([]byte, error)
	// A cluster configuration message was received
	ClusterConfig(nodeID string, config ClusterConfigMessage)
	// The peer node closed the connection
	Close(nodeID string, err error)
}

type Connection interface {
	ID() string
	Index(repo string, files []FileInfo)
	Request(repo string, name string, offset int64, size int) ([]byte, error)
	ClusterConfig(config ClusterConfigMessage)
	Statistics() Statistics
}

type rawConnection struct {
	id       string
	receiver Model
	state    int

	reader io.ReadCloser
	cr     *countingReader
	xr     *xdr.Reader

	writer io.WriteCloser
	cw     *countingWriter
	wb     *bufio.Writer
	xw     *xdr.Writer

	awaiting    []chan asyncResult
	awaitingMut sync.Mutex

	idxSent map[string]map[string]uint64
	idxMut  sync.Mutex // ensures serialization of Index calls

	nextID chan int
	outbox chan []encodable
	closed chan struct{}
	once   sync.Once

	incomingIndexes chan incomingIndex
}

type incomingIndex struct {
	update bool
	id     string
	repo   string
	files  []FileInfo
}

type asyncResult struct {
	val []byte
	err error
}

const (
	pingTimeout  = 30 * time.Second
	pingIdleTime = 60 * time.Second
)

func NewConnection(nodeID string, reader io.Reader, writer io.Writer, receiver Model) Connection {
	cr := &countingReader{Reader: reader}
	cw := &countingWriter{Writer: writer}

	flrd := flate.NewReader(cr)
	flwr, err := flate.NewWriter(cw, flate.BestSpeed)
	if err != nil {
		panic(err)
	}
	wb := bufio.NewWriter(flwr)

	c := rawConnection{
		id:              nodeID,
		receiver:        nativeModel{receiver},
		state:           stateInitial,
		reader:          flrd,
		cr:              cr,
		xr:              xdr.NewReader(flrd),
		writer:          flwr,
		cw:              cw,
		wb:              wb,
		xw:              xdr.NewWriter(wb),
		awaiting:        make([]chan asyncResult, 0x1000),
		idxSent:         make(map[string]map[string]uint64),
		outbox:          make(chan []encodable),
		nextID:          make(chan int),
		closed:          make(chan struct{}),
		incomingIndexes: make(chan incomingIndex, 100), // should be enough for anyone, right?
	}

	go c.indexSerializerLoop()
	go c.readerLoop()
	go c.writerLoop()
	go c.pingerLoop()
	go c.idGenerator()

	return wireFormatConnection{&c}
}

func (c *rawConnection) ID() string {
	return c.id
}

// Index writes the list of file information to the connected peer node
func (c *rawConnection) Index(repo string, idx []FileInfo) {
	c.idxMut.Lock()
	defer c.idxMut.Unlock()

	var msgType int
	if c.idxSent[repo] == nil {
		// This is the first time we send an index.
		msgType = messageTypeIndex

		c.idxSent[repo] = make(map[string]uint64)
		for _, f := range idx {
			c.idxSent[repo][f.Name] = f.Version
		}
	} else {
		// We have sent one full index. Only send updates now.
		msgType = messageTypeIndexUpdate
		var diff []FileInfo
		for _, f := range idx {
			if vs, ok := c.idxSent[repo][f.Name]; !ok || f.Version != vs {
				diff = append(diff, f)
				c.idxSent[repo][f.Name] = f.Version
			}
		}
		idx = diff
	}

	if msgType == messageTypeIndex || len(idx) > 0 {
		c.send(header{0, -1, msgType}, IndexMessage{repo, idx})
	}
}

// Request returns the bytes for the specified block after fetching them from the connected peer.
func (c *rawConnection) Request(repo string, name string, offset int64, size int) ([]byte, error) {
	var id int
	select {
	case id = <-c.nextID:
	case <-c.closed:
		return nil, ErrClosed
	}

	c.awaitingMut.Lock()
	if ch := c.awaiting[id]; ch != nil {
		panic("id taken")
	}
	rc := make(chan asyncResult, 1)
	c.awaiting[id] = rc
	c.awaitingMut.Unlock()

	ok := c.send(header{0, id, messageTypeRequest},
		RequestMessage{repo, name, uint64(offset), uint32(size)})
	if !ok {
		return nil, ErrClosed
	}

	res, ok := <-rc
	if !ok {
		return nil, ErrClosed
	}
	return res.val, res.err
}

// ClusterConfig send the cluster configuration message to the peer and returns any error
func (c *rawConnection) ClusterConfig(config ClusterConfigMessage) {
	c.send(header{0, -1, messageTypeClusterConfig}, config)
}

func (c *rawConnection) ping() bool {
	var id int
	select {
	case id = <-c.nextID:
	case <-c.closed:
		return false
	}

	rc := make(chan asyncResult, 1)
	c.awaitingMut.Lock()
	c.awaiting[id] = rc
	c.awaitingMut.Unlock()

	ok := c.send(header{0, id, messageTypePing})
	if !ok {
		return false
	}

	res, ok := <-rc
	return ok && res.err == nil
}

func (c *rawConnection) readerLoop() (err error) {
	defer func() {
		c.close(err)
	}()

	for {
		select {
		case <-c.closed:
			return ErrClosed
		default:
		}

		var hdr header
		hdr.decodeXDR(c.xr)
		if err := c.xr.Error(); err != nil {
			return err
		}
		if hdr.version != 0 {
			return fmt.Errorf("protocol error: %s: unknown message version %#x", c.id, hdr.version)
		}

		switch hdr.msgType {
		case messageTypeIndex:
			if c.state < stateCCRcvd {
				return fmt.Errorf("protocol error: index message in state %d", c.state)
			}
			if err := c.handleIndex(); err != nil {
				return err
			}
			c.state = stateIdxRcvd

		case messageTypeIndexUpdate:
			if c.state < stateIdxRcvd {
				return fmt.Errorf("protocol error: index update message in state %d", c.state)
			}
			if err := c.handleIndexUpdate(); err != nil {
				return err
			}

		case messageTypeRequest:
			if c.state < stateIdxRcvd {
				return fmt.Errorf("protocol error: request message in state %d", c.state)
			}
			if err := c.handleRequest(hdr); err != nil {
				return err
			}

		case messageTypeResponse:
			if c.state < stateIdxRcvd {
				return fmt.Errorf("protocol error: response message in state %d", c.state)
			}
			if err := c.handleResponse(hdr); err != nil {
				return err
			}

		case messageTypePing:
			c.send(header{0, hdr.msgID, messageTypePong})

		case messageTypePong:
			c.handlePong(hdr)

		case messageTypeClusterConfig:
			if c.state != stateInitial {
				return fmt.Errorf("protocol error: cluster config message in state %d", c.state)
			}
			if err := c.handleClusterConfig(); err != nil {
				return err
			}
			c.state = stateCCRcvd

		default:
			return fmt.Errorf("protocol error: %s: unknown message type %#x", c.id, hdr.msgType)
		}
	}
}

func (c *rawConnection) indexSerializerLoop() {
	// We must avoid blocking the reader loop when processing large indexes.
	// There is otherwise a potential deadlock where both sides has the model
	// locked because it's sending a large index update and can't receive the
	// large index update from the other side. But we must also ensure to
	// process the indexes in the order they are received, hence the separate
	// routine and buffered channel.
	for {
		select {
		case ii := <-c.incomingIndexes:
			if ii.update {
				c.receiver.IndexUpdate(ii.id, ii.repo, ii.files)
			} else {
				c.receiver.Index(ii.id, ii.repo, ii.files)
			}
		case <-c.closed:
			return
		}
	}
}

func (c *rawConnection) handleIndex() error {
	var im IndexMessage
	im.decodeXDR(c.xr)
	if err := c.xr.Error(); err != nil {
		return err
	} else {

		// We run this (and the corresponding one for update, below)
		// in a separate goroutine to avoid blocking the read loop.
		// There is otherwise a potential deadlock where both sides
		// has the model locked because it's sending a large index
		// update and can't receive the large index update from the
		// other side.

		c.incomingIndexes <- incomingIndex{false, c.id, im.Repository, im.Files}
	}
	return nil
}

func (c *rawConnection) handleIndexUpdate() error {
	var im IndexMessage
	im.decodeXDR(c.xr)
	if err := c.xr.Error(); err != nil {
		return err
	} else {
		c.incomingIndexes <- incomingIndex{true, c.id, im.Repository, im.Files}
	}
	return nil
}

func (c *rawConnection) handleRequest(hdr header) error {
	var req RequestMessage
	req.decodeXDR(c.xr)
	if err := c.xr.Error(); err != nil {
		return err
	}
	go c.processRequest(hdr.msgID, req)
	return nil
}

func (c *rawConnection) handleResponse(hdr header) error {
	data := c.xr.ReadBytesMax(256 * 1024) // Sufficiently larger than max expected block size

	if err := c.xr.Error(); err != nil {
		return err
	}

	c.awaitingMut.Lock()
	if rc := c.awaiting[hdr.msgID]; rc != nil {
		c.awaiting[hdr.msgID] = nil
		rc <- asyncResult{data, nil}
		close(rc)
	}
	c.awaitingMut.Unlock()

	return nil
}

func (c *rawConnection) handlePong(hdr header) {
	c.awaitingMut.Lock()
	if rc := c.awaiting[hdr.msgID]; rc != nil {
		c.awaiting[hdr.msgID] = nil
		rc <- asyncResult{}
		close(rc)
	}
	c.awaitingMut.Unlock()
}

func (c *rawConnection) handleClusterConfig() error {
	var cm ClusterConfigMessage
	cm.decodeXDR(c.xr)
	if err := c.xr.Error(); err != nil {
		return err
	} else {
		go c.receiver.ClusterConfig(c.id, cm)
	}
	return nil
}

type encodable interface {
	encodeXDR(*xdr.Writer) (int, error)
}
type encodableBytes []byte

func (e encodableBytes) encodeXDR(xw *xdr.Writer) (int, error) {
	return xw.WriteBytes(e)
}

func (c *rawConnection) send(h header, es ...encodable) bool {
	if h.msgID < 0 {
		select {
		case id := <-c.nextID:
			h.msgID = id
		case <-c.closed:
			return false
		}
	}
	msg := append([]encodable{h}, es...)

	select {
	case c.outbox <- msg:
		return true
	case <-c.closed:
		return false
	}
}

func (c *rawConnection) writerLoop() {
	var err error
	for {
		select {
		case es := <-c.outbox:
			for _, e := range es {
				e.encodeXDR(c.xw)
			}

			if err = c.flush(); err != nil {
				c.close(err)
				return
			}
		case <-c.closed:
			return
		}
	}
}

type flusher interface {
	Flush() error
}

func (c *rawConnection) flush() error {
	if err := c.xw.Error(); err != nil {
		return err
	}

	if err := c.wb.Flush(); err != nil {
		return err
	}

	if f, ok := c.writer.(flusher); ok {
		return f.Flush()
	}

	return nil
}

func (c *rawConnection) close(err error) {
	c.once.Do(func() {
		close(c.closed)

		c.awaitingMut.Lock()
		for i, ch := range c.awaiting {
			if ch != nil {
				close(ch)
				c.awaiting[i] = nil
			}
		}
		c.awaitingMut.Unlock()

		go c.receiver.Close(c.id, err)
	})
}

func (c *rawConnection) idGenerator() {
	nextID := 0
	for {
		nextID = (nextID + 1) & 0xfff
		select {
		case c.nextID <- nextID:
		case <-c.closed:
			return
		}
	}
}

func (c *rawConnection) pingerLoop() {
	var rc = make(chan bool, 1)
	ticker := time.Tick(pingIdleTime / 2)
	for {
		select {
		case <-ticker:
			if d := time.Since(c.xr.LastRead()); d < pingIdleTime {
				if debug {
					l.Debugln(c.id, "ping skipped after rd", d)
				}
				continue
			}
			if d := time.Since(c.xw.LastWrite()); d < pingIdleTime {
				if debug {
					l.Debugln(c.id, "ping skipped after wr", d)
				}
				continue
			}
			go func() {
				if debug {
					l.Debugln(c.id, "ping ->")
				}
				rc <- c.ping()
			}()
			select {
			case ok := <-rc:
				if debug {
					l.Debugln(c.id, "<- pong")
				}
				if !ok {
					c.close(fmt.Errorf("ping failure"))
				}
			case <-time.After(pingTimeout):
				c.close(fmt.Errorf("ping timeout"))
			case <-c.closed:
				return
			}

		case <-c.closed:
			return
		}
	}
}

func (c *rawConnection) processRequest(msgID int, req RequestMessage) {
	data, _ := c.receiver.Request(c.id, req.Repository, req.Name, int64(req.Offset), int(req.Size))

	c.send(header{0, msgID, messageTypeResponse}, encodableBytes(data))
}

type Statistics struct {
	At            time.Time
	InBytesTotal  uint64
	OutBytesTotal uint64
}

func (c *rawConnection) Statistics() Statistics {
	return Statistics{
		At:            time.Now(),
		InBytesTotal:  c.cr.Tot(),
		OutBytesTotal: c.cw.Tot(),
	}
}

func IsDeleted(bits uint32) bool {
	return bits&FlagDeleted != 0
}

func IsInvalid(bits uint32) bool {
	return bits&FlagInvalid != 0
}

func IsDirectory(bits uint32) bool {
	return bits&FlagDirectory != 0
}

func HasPermissionBits(bits uint32) bool {
	return bits&FlagNoPermBits == 0
}