struct.go

package byzcoin

import (
	"bytes"
	"encoding/binary"
	"fmt"
	"go.dedis.ch/onet/v3/log"
	"go.dedis.ch/onet/v3/network"
	"sort"
	"strings"
	"sync"
	"sync/atomic"
	"time"

	"go.dedis.ch/cothority/v3"
	"go.dedis.ch/cothority/v3/skipchain"
	"go.dedis.ch/onet/v3"
	"go.dedis.ch/protobuf"
	bbolt "go.etcd.io/bbolt"
	"golang.org/x/xerrors"
)

const defaultMaxSize = (1 << 31) - 1 // maximum 32-bit int
const versionLength = 64 / 8         // bytes
const prefixLength = 32              // bytes
const cleanThreshold = 0.8

// number of blocks in the queue before the service doesn't notify new
// blocks anymore.
// This is per-queue, so if one queue doesn't collect its blocks,
// the other queues continue normally.
const notificationQueueLenght = 8

var bucketStateChangeStorage = []byte("statechangestorage")
var errLengthInstanceID = xerrors.New("InstanceID must have 32 bytes")

// StateChangeEntry is the object stored to keep track of instance history. It
// contains the state change and the block index
type StateChangeEntry struct {
	StateChange StateChange
	TxIndex     int
	BlockIndex  int
	Timestamp   time.Time
}

// StateChangeEntries is an array of StateChangeEntry and can be
// sorted to restore the transaction order
type StateChangeEntries []StateChangeEntry

func (sces StateChangeEntries) Len() int {
	return len(sces)
}

func (sces StateChangeEntries) Less(i, j int) bool {
	return sces[i].TxIndex < sces[j].TxIndex
}

func (sces StateChangeEntries) Swap(i, j int) {
	sces[i], sces[j] = sces[j], sces[i]
}

// Copy creates a deep copy of the statechange, so that tests
// can correctly work on those copies.
func (sc StateChange) Copy() StateChange {
	c := StateChange{
		StateAction: sc.StateAction,
		Version:     sc.Version,
	}
	c.InstanceID = append([]byte{}, sc.InstanceID...)
	c.ContractID = sc.ContractID
	c.Value = append([]byte{}, sc.Value...)
	c.DarcID = append([]byte{}, sc.DarcID...)
	return c
}

// stateChangeStorage stores the state changes using their instance ID, the block index and
// their version to yield a key. This key has the property to sort the key-value pairs
// first by instance ID and then by version so we can use the BoltDB key traversal.
// The block index is appended only to access more efficiently to the information
// without having to decode the value.
// The storage cleans up by itself with respect to the parameters when appending new
// state changes. If the size goes above the limit, each skipchain is truncated by its
// oldest block until the space threshold is reached.
type stateChangeStorage struct {
	db *bbolt.DB
	sync.Mutex
	bucket      []byte
	size        int
	maxSize     int
	maxNbrBlock int
}

// Create a storage with a default maximum size
func newStateChangeStorage(c *onet.Context) *stateChangeStorage {
	db, name := c.GetAdditionalBucket(bucketStateChangeStorage)
	return &stateChangeStorage{
		db:      db,
		bucket:  name,
		maxSize: defaultMaxSize,
	}
}

// getBucket gets the bucket for the given skipchain
func (s *stateChangeStorage) getBucket(tx *bbolt.Tx, sid skipchain.SkipBlockID) *bbolt.Bucket {
	b := tx.Bucket(s.bucket)
	if b == nil {
		panic("Bucket has not been created. This is a programmer error.")
	}

	if tx.Writable() {
		sbb, err := b.CreateBucketIfNotExists(sid)
		if err != nil {
			panic(err)
		}

		return sbb
	}

	return b.Bucket(sid)
}

// setMaxSize enables the cleaning of old state changes when the storage
// size is above a given threshold. Note that the value is not strict.
func (s *stateChangeStorage) setMaxSize(size int) {
	s.maxSize = size
}

// calculateSize reads the entries in the database and sums up their
// sizes
func (s *stateChangeStorage) calculateSize() error {
	dSize := 0
	err := s.db.View(func(tx *bbolt.Tx) error {
		b := tx.Bucket(s.bucket)
		if b == nil {
			return xerrors.New("Missing bucket")
		}

		return b.ForEach(func(scid, v []byte) error {
			scb := b.Bucket(scid)
			if scb == nil {
				return nil
			}

			return scb.ForEach(func(k, v []byte) error {
				dSize += len(v)
				return nil
			})
		})
	})
	if err != nil {
		return fmt.Errorf("couldn't calculate size: %v", err)
	}

	s.Lock()
	s.size += dSize
	s.Unlock()

	return nil
}

// This will clean the oldest state changes when the total size
// is above the maximum. It will remove elements until cleanThreshold of
// the space is available.
func (s *stateChangeStorage) cleanBySize() error {
	if s.size < s.maxSize || s.maxSize == 0 {
		// nothing to clean
		return nil
	}

	size := s.size

	err := s.db.Update(func(tx *bbolt.Tx) error {
		// We make enough space to not have to do it everytime
		// we append state changes
		thres := int(float64(s.maxSize) * cleanThreshold)

		b := tx.Bucket(s.bucket)
		if b == nil {
			return xerrors.New("Missing bucket")
		}

		// loop until enough blocks have been cleaned
		for size > thres {
			// Each pair at this level is a bucket assigned to a skipchain
			err := b.ForEach(func(scid, _ []byte) error {
				scb := b.Bucket(scid)
				if scb == nil {
					return nil
				}

				// we first look for the oldest block for the skipchain
				oldestIndex := int64(-1)
				var idx int64
				c := scb.Cursor()
				k, _ := c.First()
				for k != nil {
					buf := bytes.NewBuffer(k[prefixLength+versionLength:])

					err := binary.Read(buf, binary.BigEndian, &idx)
					if err != nil {
						return xerrors.Errorf("decoding index: %v", err)
					}

					if oldestIndex == -1 || oldestIndex > idx {
						oldestIndex = idx
					}

					// go to the next instance
					k, _ = c.Seek(s.keyOfLast(k[:prefixLength]))
				}

				// ... and we clean it
				k, v := c.First()
				for k != nil {
					buf := bytes.NewBuffer(k[prefixLength+versionLength:])

					err := binary.Read(buf, binary.BigEndian, &idx)
					if err != nil {
						return xerrors.Errorf("decoding index: %v", err)
					}

					if oldestIndex == idx {
						if err := c.Delete(); err != nil {
							return xerrors.Errorf("deleting pair: %v", err)
						}

						size -= len(v)
						k, v = c.Next()
					} else {
						// jump to the next instance as entries are ordered by version (thus by block)
						k, v = c.Seek(s.keyOfLast(k[:prefixLength]))
					}
				}

				if scb.Stats().KeyN == 0 {
					if err := b.DeleteBucket(scid); err != nil {
						return xerrors.Errorf("deleting bucket: %v", err)
					}
				}

				return nil
			})

			if err != nil {
				return xerrors.Errorf("processing pairs: %v", err)
			}
		}

		return nil
	})

	if err != nil {
		return xerrors.Errorf("tx error: %v", err)
	}

	s.size = size
	return nil
}

// This will clean the state changes per instance where the block
// index is too low compared to the threshold
func (s *stateChangeStorage) cleanByBlock(scs StateChanges, sb *skipchain.SkipBlock) error {
	if s.maxNbrBlock == 0 {
		return nil
	}

	thres := int64(sb.Index - s.maxNbrBlock)
	size := s.size

	err := s.db.Update(func(tx *bbolt.Tx) error {
		b := s.getBucket(tx, sb.SkipChainID())

		// Prevent from cleaning the same instance twice
		done := map[string]bool{}

		// Clean only the instances where state changes have been added
		for _, sc := range scs {
			_, ok := done[string(sc.InstanceID)]
			if !ok {
				done[string(sc.InstanceID)] = true

				var buf bytes.Buffer
				// The key is built using BigEndian order
				err := binary.Write(&buf, binary.BigEndian, thres)
				if err != nil {
					return xerrors.Errorf("encoding threshold: %v", err)
				}
				index := buf.Bytes()

				c := b.Cursor()
				for k, v := c.Seek(sc.InstanceID); k != nil && bytes.HasPrefix(k, sc.InstanceID); k, v = c.Next() {
					if bytes.Compare(k[len(k)-len(index):], index) <= 0 {
						if err := c.Delete(); err != nil {
							return xerrors.Errorf("deleting item: %v", err)
						}
						size -= len(v)
					}
				}
			}
		}

		return nil
	})

	if err == nil {
		s.size = size
	}

	return cothority.ErrorOrNil(err, "tx error")
}

// this generates a storage key using the instance ID and the version
func (s *stateChangeStorage) key(iid []byte, ver uint64, idx int64) ([]byte, error) {
	b := bytes.Buffer{}
	_, err := b.Write(iid)
	if err != nil {
		return nil, xerrors.Errorf("writing id: %v", err)
	}

	// BigEndian is used here because of the byte-sorted order of
	// BoltDB when iterating over the keys
	err = binary.Write(&b, binary.BigEndian, ver)
	if err != nil {
		return nil, xerrors.Errorf("writing version: %v", err)
	}

	err = binary.Write(&b, binary.BigEndian, idx)
	if err != nil {
		return nil, xerrors.Errorf("writing index: %v", err)
	}

	return b.Bytes(), nil
}

// Takes an instance ID and returns the last possible key for it which can be used
// to go the next instance first key
func (s *stateChangeStorage) keyOfLast(iid []byte) []byte {
	key := make([]byte, len(iid))
	copy(key, iid)
	return append(key, []byte{0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff}...)
}

// this will clean the oldest state changes until there is enough
// space left and append the new ones
func (s *stateChangeStorage) append(scs StateChanges, sb *skipchain.SkipBlock) error {
	s.Lock()
	defer s.Unlock()
	// Run a cleaning procedure first to insure we're not above the limit
	err := s.cleanBySize()
	if err != nil {
		return xerrors.Errorf("error when cleaning: %v", err)
	}

	size := s.size

	err = s.db.Update(func(tx *bbolt.Tx) error {
		b := s.getBucket(tx, sb.SkipChainID())

		// append each list of state changes (or create the entry)
		for i, sc := range scs {
			if len(sc.InstanceID) != prefixLength {
				// as we use it as a prefix, all must have the same length
				return cothority.WrapError(errLengthInstanceID)
			}

			key, err := s.key(sc.InstanceID, sc.Version, int64(sb.Index))
			if err != nil {
				return xerrors.Errorf("key: %v", err)
			}
			if b.Get(key) != nil {
				return fmt.Errorf("already existing scs: %x - %d / %d",
					sc.InstanceID, sc.Version, sb.Index)
			}

			now := time.Now()
			buf, err := protobuf.Encode(&StateChangeEntry{
				StateChange: sc,
				TxIndex:     i,
				BlockIndex:  sb.Index,
				Timestamp:   now,
			})
			if err != nil {
				return xerrors.Errorf("encoding: %v", err)
			}

			err = b.Put(key, buf)
			if err != nil {
				return xerrors.Errorf("writing item: %v", err)
			}

			size += len(buf)
		}

		return nil
	})
	if err != nil {
		return xerrors.Errorf("tx error: %v", err)
	}

	s.size = size

	return cothority.ErrorOrNil(s.cleanByBlock(scs, sb), "cleaning")
}

// This will return the list of state changes for the given instance
func (s *stateChangeStorage) getAll(iid []byte, sid skipchain.SkipBlockID) (entries []StateChangeEntry, err error) {
	s.Lock()
	defer s.Unlock()
	if len(iid) != prefixLength {
		return nil, cothority.WrapError(errLengthInstanceID)
	}

	err = s.db.View(func(tx *bbolt.Tx) error {
		b := s.getBucket(tx, sid)
		if b == nil {
			// Nothing yet stored for this instance
			return nil
		}

		c := b.Cursor()
		for k, v := c.Seek(iid); bytes.HasPrefix(k, iid); k, v = c.Next() {
			var sce StateChangeEntry
			err = protobuf.Decode(v, &sce)
			if err != nil {
				return xerrors.Errorf("decoding: %v", err)
			}

			entries = append(entries, sce)
		}

		return nil
	})

	err = cothority.ErrorOrNil(err, "tx error")
	return
}

// This will return the state change entry for the given instance and version.
// Use the bool returned value to check if the version exists
func (s *stateChangeStorage) getByVersion(iid []byte,
	ver uint64, sid skipchain.SkipBlockID) (sce StateChangeEntry, ok bool, err error) {
	s.Lock()
	defer s.Unlock()
	if len(iid) != prefixLength {
		err = cothority.WrapError(errLengthInstanceID)
		return
	}

	key, err := s.key(iid, ver, int64(0))
	if err != nil {
		err = xerrors.Errorf("key: %v", err)
		return
	}

	// Only the instance ID and the version is known
	prefix := key[:prefixLength+versionLength]

	err = s.db.View(func(tx *bbolt.Tx) error {
		b := s.getBucket(tx, sid)

		c := b.Cursor()
		k, v := c.Seek(prefix)
		if k != nil && bytes.HasPrefix(k, prefix) {
			err := protobuf.Decode(v, &sce)
			if err != nil {
				return xerrors.Errorf("decoding: %v", err)
			}

			ok = true
		}

		return nil
	})

	err = cothority.ErrorOrNil(err, "tx error")
	return
}

// getByBlock looks for the state changes associated with a given
// skipblock
func (s *stateChangeStorage) getByBlock(sid skipchain.SkipBlockID, idx int) (entries StateChangeEntries, err error) {
	s.Lock()
	defer s.Unlock()
	err = s.db.View(func(tx *bbolt.Tx) error {
		b := s.getBucket(tx, sid)
		if b == nil {
			// No bucket means that the chain hasn't been processed yet.
			return nil
		}

		var suffix bytes.Buffer
		// The key is built using BigEndian order
		binary.Write(&suffix, binary.BigEndian, int64(idx))

		c := b.Cursor()
		for k, v := c.First(); k != nil; k, v = c.Next() {
			if bytes.HasSuffix(k, suffix.Bytes()) {
				var sce StateChangeEntry
				err = protobuf.Decode(v, &sce)
				if err != nil {
					return xerrors.Errorf("decoding: %v", err)
				}

				entries = append(entries, sce)
			}
		}

		return nil
	})

	sort.Sort(entries)
	err = cothority.ErrorOrNil(err, "tx error")
	return
}

// getLast looks for the last version of a given instance and return the entry. Use
// the bool value to know if there is a hit or not.
func (s *stateChangeStorage) getLast(iid []byte, sid skipchain.SkipBlockID) (sce StateChangeEntry, ok bool, err error) {
	s.Lock()
	defer s.Unlock()
	if len(iid) != prefixLength {
		err = cothority.WrapError(errLengthInstanceID)
		return
	}

	err = s.db.View(func(tx *bbolt.Tx) error {
		b := s.getBucket(tx, sid)
		c := b.Cursor()

		// Seek the next instance ID and take a step back
		// to reach the newest version.
		// By appending 2^64-1 to the key, we get the last
		// possible key.
		c.Seek(s.keyOfLast(iid))
		k, v := c.Prev()

		if bytes.HasPrefix(k, iid) {
			err := protobuf.Decode(v, &sce)
			if err != nil {
				return xerrors.Errorf("decoding: %v", err)
			}

			ok = true
		}
		return nil
	})

	err = cothority.ErrorOrNil(err, "tx error")
	return
}

// SafeAdd will add a to the value of the coin if there will be no
// overflow.
func (c *Coin) SafeAdd(a uint64) error {
	s1 := c.Value + a
	if s1 < c.Value || s1 < a {
		return xerrors.New("uint64 overflow")
	}
	c.Value = s1
	return nil
}

// SafeSub subtracts a from the value of the coin if there
// will be no underflow.
func (c *Coin) SafeSub(a uint64) error {
	if a <= c.Value {
		c.Value -= a
		return nil
	}
	return xerrors.New("uint64 underflow")
}

// SafeTransfer takes val from one coin and puts it to another.
// It checks that the source coin has enough,
// and that the destination coin will not overflow.
func (c *Coin) SafeTransfer(to *Coin, val uint64) error {
	if err := c.SafeSub(val); err != nil {
		return err
	}
	if err := to.SafeAdd(val); err != nil {
		_ = c.SafeAdd(val)
		return err
	}
	return nil
}

type notification struct {
	block  *skipchain.SkipBlock
	txs    TxResults
	hashes [][]byte
}

func (n *notification) getTx(id []byte) *TxResult {
	for i, h := range n.hashes {
		if bytes.Equal(h, id) {
			return &n.txs[i]
		}
	}

	return nil
}

type waitChannel = chan *notification

type bcNotifications struct {
	sync.Mutex
	// blockListeners will be notified every time a block is created.
	// It is up to them to filter out block creations on chains they are not
	// interested in.
	blockListeners []waitChannel
}

func (bc *bcNotifications) informBlock(block *skipchain.SkipBlock, txs TxResults) {
	hashes := make([][]byte, len(txs))
	for i, tx := range txs {
		// Pre-computed hash to save some computation load.
		hashes[i] = tx.ClientTransaction.Instructions.Hash()
	}

	notif := &notification{
		block:  block,
		txs:    txs,
		hashes: hashes,
	}

	bc.Lock()
	defer bc.Unlock()
	for _, x := range bc.blockListeners {
		select {
		case x <- notif:
		default:
			log.Warn("not queueing up block for notification because queue is" +
				" full")
		}
	}
}

func (bc *bcNotifications) registerForBlocks() waitChannel {
	bc.Lock()
	defer bc.Unlock()

	ch := make(waitChannel, notificationQueueLenght)
	bc.blockListeners = append(bc.blockListeners, ch)

	return ch
}

func (bc *bcNotifications) unregisterForBlocks(ch waitChannel) {
	bc.Lock()
	defer bc.Unlock()

	for i, listener := range bc.blockListeners {
		if listener == ch {
			bc.blockListeners = append(bc.blockListeners[0:i], bc.blockListeners[i+1:]...)
		}
	}
}

func (c ChainConfig) sanityCheck(old *ChainConfig, version Version) error {
	if c.BlockInterval <= 0 {
		return xerrors.New("block interval is less or equal to zero")
	}
	// too small would make it impossible to even send through a config update tx to fix it,
	// so don't allow that.
	if c.MaxBlockSize < 16000 {
		return xerrors.New("max block size is less than 16000")
	}
	// onet/network.MaxPacketSize is 10 megs, leave some headroom anyway.
	if c.MaxBlockSize > 8*1e6 {
		return xerrors.New("max block size is greater than 8 megs")
	}
	if len(c.Roster.List) < 3 {
		return xerrors.New("need at least 3 nodes to have a majority")
	}

	if version >= VersionRosterCheck {
		for i, si := range c.Roster.List {
			if err := c.nodeCheck(i, si); err != nil {
				return xerrors.Errorf("while checking roster: %v", err)
			}
		}
	}

	if old != nil {
		return cothority.ErrorOrNil(old.checkNewRoster(c.Roster), "roster check")
	}
	return nil
}

func (c ChainConfig) nodeCheck(i int, si *network.ServerIdentity) error {
	err := fmt.Sprintf("node[%d] = %s of roster", i, si.Description)
	if si.Public == nil {
		return xerrors.Errorf("%s has no public key", err)
	}
	if si.Address == "" {
		return xerrors.Errorf("%s has no address", err)
	}
	if si.Description == "" {
		return xerrors.Errorf("%s has no description", err)
	}
searchService:
	for _, service := range []string{ServiceName,
		skipchain.ServiceName} {
		for _, si := range si.ServiceIdentities {
			if si.Name == service && si.Public != nil && si.Suite != "" {
				continue searchService
			}
		}
		return xerrors.Errorf("%s is missing service %s", service)
	}
	return nil
}

// checkNewRoster makes sure that the new roster follows the rules we need
// in byzcoin:
//   - no new node can join as leader
//   - only one node joining or leaving
func (c ChainConfig) checkNewRoster(newRoster onet.Roster) error {
	// Check new leader was in old roster
	if index, _ := c.Roster.Search(newRoster.List[0].ID); index < 0 {
		return xerrors.New("new leader must be in previous roster")
	}

	// Check we don't change more than one node
	added := 0
	oldList := onet.NewRoster(c.Roster.List)
	for _, si := range newRoster.List {
		if i, _ := oldList.Search(si.ID); i >= 0 {
			oldList.List = append(oldList.List[:i], oldList.List[i+1:]...)
		} else {
			added++
		}
	}
	if len(oldList.List)+added > 1 {
		return xerrors.New("can only change one node at a time - adding or removing")
	}
	return nil
}

// String implements a nicer text representation of a Chainconfig.
//
// Here is an example of what it outputs:
//
// ```
// - ChainConfig:
// -- BlockInterval: 7s
// -- Roster: {1e89775c-636a-536a-bc39-1ec951c86dc9 [tls:https://localhost:2002 tls:https://localhost:2004 tls:https://localhost:2006] 467abd382f78222e898d323194c0fb30f7096bb6bb885ea31284979f794e558a}
// -- MaxBlockSize: 5000000
// -- DarcContractIDs:
// --- darc contract ID 0: darc
// --- darc contract ID 1: darc2
// --- darc contract ID 2: darc3'
// ```
func (c ChainConfig) String() string {
	res := new(strings.Builder)
	res.WriteString("- ChainConfig:\n")
	fmt.Fprintf(res, "-- BlockInterval: %s\n", c.BlockInterval.String())
	fmt.Fprintf(res, "-- Roster: %s\n", c.Roster)
	fmt.Fprintf(res, "-- MaxBlockSize: %d\n", c.MaxBlockSize)
	res.WriteString("-- DarcContractIDs:\n")
	for i, darcID := range c.DarcContractIDs {
		fmt.Fprintf(res, "--- darc contract ID %d: %s\n", i, darcID)
	}
	return res.String()
}

// GetUpdatesFlags define how the proofs will be returned
type GetUpdatesFlags uint64

const (
	// GUFSendVersion0 will make GetUpdates to send all instances with
	// version 0, even those that are note updated.
	GUFSendVersion0 = GetUpdatesFlags(1 << iota)
	// GUFSendMissingProofs will make GetUpdates send proofs for missing
	// instances. If not present, missing instances are ignored.
	GUFSendMissingProofs
)

// runSingleWG allows to ensure a single execution of a method.
type runSingleWG struct {
	wg      sync.WaitGroup
	running int32
}

// Increment on Start - returns false if wg is already running
func (cg *runSingleWG) start() bool {
	fresh := atomic.CompareAndSwapInt32(&cg.running, 0, 1)
	if !fresh {
		return false
	}
	cg.wg.Add(1)
	return true
}

// Decrement on Done
func (cg *runSingleWG) done() {
	if !cg.isRunning() {
		panic("cannot call done on idle runSingleWG")
	}
	atomic.AddInt32(&cg.running, -1)
	cg.wg.Done()
}

// Check if currently running
func (cg *runSingleWG) isRunning() bool {
	return atomic.LoadInt32(&cg.running) > 0
}

// wait for the single task to be done
func (cg *runSingleWG) wait() {
	cg.wg.Wait()
}

// tasksWG is a special workGroup that can be paused or resumed.
// The default state is paused.
type tasksWG struct {
	m            sync.Mutex
	wg           sync.WaitGroup
	tasksAllowed bool
}

// Add returns false if tasksWG is paused, and the WaitGroup is not increased.
// Only if tasksWG is resumed Add returns true and the WaitGroup is
// increased.
func (wwg *tasksWG) add(delta int) bool {
	wwg.m.Lock()
	defer wwg.m.Unlock()
	if !wwg.tasksAllowed {
		return false
	}
	wwg.wg.Add(delta)
	return true
}

// pauses tasksWG so that no new tasks are allowed.
func (wwg *tasksWG) pause() bool {
	wwg.m.Lock()
	defer wwg.m.Unlock()
	if !wwg.tasksAllowed {
		return false
	}
	wwg.tasksAllowed = false
	return true
}

// resumes tasksWG to allow new tasks again.
func (wwg *tasksWG) resume() {
	wwg.m.Lock()
	defer wwg.m.Unlock()
	if wwg.tasksAllowed {
		panic("cannot resume when already resumed")
	}
	wwg.tasksAllowed = true
}

// areTasksAllowed returns the state of tasksWG.
func (wwg *tasksWG) areTasksAllowed() bool {
	wwg.m.Lock()
	defer wwg.m.Unlock()
	return wwg.tasksAllowed
}

// finish one of the tasks
func (wwg *tasksWG) done() {
	wwg.wg.Done()
}

// wait for all the tasks to be finished
func (wwg *tasksWG) wait() {
	wwg.wg.Wait()
}