// Copyright 2015 The Vanadium Authors. All rights reserved.
// Use of this source code is governed by a BSD-style
// license that can be found in the LICENSE file.

package vsync

// Initiator requests deltas from a chosen peer for all the syncgroups in common
// across all apps/databases. It then modifies the sync metadata (DAG and local
// log records) based on the deltas, detects and resolves conflicts if any, and
// suitably updates the local Databases.

import (
	"reflect"
	"sort"
	"strings"
	"time"

	"v.io/v23"
	"v.io/v23/context"
	"v.io/v23/naming"
	"v.io/v23/options"
	"v.io/v23/security"
	"v.io/v23/security/access"
	"v.io/v23/verror"
	"v.io/v23/vom"
	"v.io/x/lib/vlog"
	"v.io/x/ref/services/syncbase/server/interfaces"
	"v.io/x/ref/services/syncbase/server/util"
	"v.io/x/ref/services/syncbase/server/watchable"
	"v.io/x/ref/services/syncbase/store"
)

// getDeltasFromPeer performs an initiation round once per database to the specified
// peer. An initiation round consists of an identification and filtering step
// and two sync rounds:
// * Sync syncgroup metadata.
// * Sync data.
//
// In the identification step, the initiator connects to the peer to identify it
// and obtain its list of blessings. It then uses these blessings to filter the
// list of syncgroups that are in common with the peer (learned based on the
// joiner list) to only include those syncgroups whose acl is satisfied by the
// peer's blessings. These syncgroups are included in the next two sync
// rounds. In the next two sync rounds, the initiator ensures that the peer
// identification (blessings) used in the first step has not changed.
//
// Note that alternately we could have used the blessings a peer uses to join
// the syncgroup to identify and validate it during sync rounds. However, this
// tight coupling prevents a peer who receives another valid blessing that
// satisfies the syncgroup acl after joining but loses the one used during join,
// from syncing.
//
// Each sync round involves:
// * Contacting the peer to receive all the deltas based on the local genvectors.
// * Processing those deltas to discover objects which have been updated.
// * Processing updated objects to detect and resolve any conflicts if needed.
// * Communicating relevant object updates to the Database in case of data.
// * Updating local genvectors to catch up to the received remote genvectors.
//
// The processing of the deltas is done one Database at a time, encompassing all
// the syncgroups common to the initiator and the responder. If a local error is
// encountered during the processing of a Database, that Database is skipped and
// the initiator continues on to the next one. If the connection to the peer
// encounters an error, this initiation round is aborted. Note that until the
// local genvectors are updated based on the received deltas (the last step in an
// initiation round), the work done by the initiator is idempotent.
//
// TODO(hpucha): Check the idempotence, esp in addNode in DAG.
func (s *syncService) getDeltasFromPeer(ctx *context.T, peer connInfo) error {
	vlog.VI(2).Infof("sync: getDeltasFromPeer: begin: contacting peer %v", peer)
	defer vlog.VI(2).Infof("sync: getDeltasFromPeer: end: contacting peer %v", peer)

	var errFinal error // the last non-nil error encountered is returned to the caller.

	info := s.copyMemberInfo(ctx, peer.relName)
	if info == nil {
		vlog.VI(4).Infof("sync: getDeltasFromPeer: copyMemberInfo failed %v", peer)
		return verror.New(verror.ErrInternal, ctx, peer.relName, "no member info found")
	}

	// Sync each Database that may have syncgroups common with this peer,
	// one at a time.
	for gdbName, dbInfo := range info.db2sg {
		if len(peer.mtTbls) < 1 && len(peer.addrs) < 1 {
			vlog.Errorf("sync: getDeltasFromPeer: no mount tables or endpoints found to connect to peer %v", peer)
			return verror.New(verror.ErrInternal, ctx, peer.relName, peer.addrs, "all mount tables failed")
		}

		var err error
		peer, err = s.getDBDeltas(ctx, gdbName, dbInfo, peer)

		if verror.ErrorID(err) == interfaces.ErrConnFail.ID {
			return err
		}

		if err != nil {
			errFinal = err
		}
	}
	return errFinal
}

// getDBDeltas performs an initiation round for the specified database.
func (s *syncService) getDBDeltas(ctx *context.T, gdbName string, info sgMemberInfo, peer connInfo) (connInfo, error) {
	vlog.VI(2).Infof("sync: getDBDeltas: begin: contacting peer %v for db %s", peer, gdbName)
	defer vlog.VI(2).Infof("sync: getDBDeltas: end: contacting peer %v for db %s", peer, gdbName)

	// Note that the "identify" step is done once per database for privacy
	// reasons. When the app blesses Syncbase during syncgroup create/join,
	// these blessings would be associated with that database, and should
	// not be revealed when that database is not being synced.
	//
	// TODO(hpucha): Revisit which blessings are sent.
	blessingNames, updPeer, err := s.identifyPeer(ctx, peer)
	if err != nil {
		return updPeer, err
	}

	// The initiation config is shared across the sync rounds in an
	// initiation. The mount tables in the config are pruned to eliminate
	// the ones that failed to reach the peer during each RPC call. This
	// helps the next RPC call to use the currently successful or the
	// not-yet tried mount tables to reach a peer, instead of retrying the
	// already failed ones.
	//
	// TODO(hpucha): Clean up sharing of the initiationConfig.
	c, err := newInitiationConfig(ctx, s, gdbName, info, updPeer)
	if err != nil {
		return updPeer, err
	}

	// Filter the syncgroups using the peer blessings, and populate the
	// syncgroup ids, the syncgroup prefixes and the remote peer blessings
	// to be used in the next two rounds of sync.
	if err := s.filterSyncgroups(ctx, c, blessingNames); err != nil {
		return c.peer, err
	}

	// Sync syncgroup metadata.
	if err = s.getDeltas(ctx, c, true); err != nil {
		// If syncgroup sync fails, abort data sync as well.
		return c.peer, err
	}

	// Sync data.
	err = s.getDeltas(ctx, c, false)
	return c.peer, err
}

// identifyPeer obtains the blessings from the remote peer.
func (s *syncService) identifyPeer(ctxIn *context.T, peer connInfo) ([]string, connInfo, error) {
	vlog.VI(2).Infof("sync: identifyPeer: begin for peer %v", peer)
	defer vlog.VI(2).Infof("sync: identifyPeer: end for peer %v", peer)

	ctx, cancel := context.WithCancel(ctxIn)
	defer cancel()

	var blessingNames []string
	rpcArgs := []interface{}{nil, peer.relName}
	op := func(ctx *context.T, peer string) (interface{}, error) {
		call, err := v23.GetClient(ctx).StartCall(ctx, peer, "GetDeltas", rpcArgs)
		if err != nil {
			return nil, err
		}
		blessingNames, _ = call.RemoteBlessings()
		call.Finish()
		return nil, nil
	}
	updPeer, _, err := runAtPeer(ctx, peer, op)
	return blessingNames, updPeer, err
}

// filterSyncgroups uses the remote peer's blessings to obtain the set of
// syncgroups that are allowed to be synced with it based on its current
// syncgroup acls.
func (s *syncService) filterSyncgroups(ctx *context.T, c *initiationConfig, blessingNames []string) error {
	vlog.VI(2).Infof("sync: filterSyncGroups: begin")
	defer vlog.VI(2).Infof("sync: filterSyncGroups: end")

	// Perform authorization.
	if len(blessingNames) == 0 {
		return verror.New(verror.ErrNoAccess, ctx)
	}

	vlog.VI(4).Infof("sync: filterSyncGroups: got peer names %v", blessingNames)

	remSgIds := make(sgSet)

	// Prepare the syncgroup prefixes since we are going through each
	// syncgroup. This is an optimization done to avoid looping and reading
	// syncgroup data another time.
	c.sgPfxs = make(map[string]sgSet)

	for gid := range c.sgIds {
		var sg *interfaces.Syncgroup
		sg, err := getSyncgroupById(ctx, c.st, gid)
		if err != nil {
			continue
		}

		if _, ok := sg.Joiners[c.peer.relName]; !ok {
			// Peer is no longer part of the syncgroup.
			continue
		}

		// TODO(hpucha): Is using the Read tag to authorize the remote
		// peer ok? The thinking here is that the peer uses the read tag
		// on the GetDeltas RPC to authorize the initiator and this
		// makes it symmetric.
		if err := authorizeForTag(ctx, sg.Spec.Perms, access.Read, blessingNames); err != nil {
			vlog.VI(4).Infof("sync: filterSyncGroups: skipping sg %v", gid)
			continue
		}

		remSgIds[gid] = struct{}{}
		for _, p := range sg.Spec.Prefixes {
			pfxStr := toTableRowPrefixStr(p)
			sgs, ok := c.sgPfxs[pfxStr]
			if !ok {
				sgs = make(sgSet)
				c.sgPfxs[pfxStr] = sgs
			}
			sgs[gid] = struct{}{}
		}
	}

	c.sgIds = remSgIds

	if len(c.sgIds) == 0 {
		return verror.New(verror.ErrInternal, ctx, "no syncgroups found after filtering", c.peer.relName, c.appName, c.dbName)
	}

	if len(c.sgPfxs) == 0 {
		return verror.New(verror.ErrInternal, ctx, "no syncgroup prefixes found", c.peer.relName, c.appName, c.dbName)
	}

	sort.Strings(blessingNames)
	c.auth = &namesAuthorizer{blessingNames}
	return nil
}

// getDeltas gets the deltas from the chosen peer. If sg flag is set to true, it
// will sync syncgroup metadata. If sg flag is false, it will sync data.
func (s *syncService) getDeltas(ctxIn *context.T, c *initiationConfig, sg bool) error {
	vlog.VI(2).Infof("sync: getDeltas: begin: contacting peer sg %v %v", sg, c.peer)
	defer vlog.VI(2).Infof("sync: getDeltas: end: contacting peer sg %v %v", sg, c.peer)

	ctx, cancel := context.WithCancel(ctxIn)
	// cancel() is idempotent.
	defer cancel()

	// Initialize initiation state for syncing this Database.
	iSt := newInitiationState(ctx, c, sg)

	if sg {
		// Create local genvecs so that they contain knowledge about
		// common syncgroups and then send the syncgroup metadata sync
		// request.
		if err := iSt.prepareSGDeltaReq(ctx); err != nil {
			return err
		}
	} else {
		// Create local genvecs so that they contain knowledge only about common
		// prefixes and then send the data sync request.
		if err := iSt.prepareDataDeltaReq(ctx); err != nil {
			return err
		}
	}

	op := func(ctx *context.T, peer string) (interface{}, error) {
		c := interfaces.SyncClient(peer)
		var err error
		// The authorizer passed to the RPC ensures that the remote
		// peer's blessing names are the same as obtained in the
		// identification step done previously.
		iSt.stream, err = c.GetDeltas(ctx, iSt.req, iSt.config.sync.name,
			options.ServerAuthorizer{iSt.config.auth}, options.ChannelTimeout(connectionTimeOut))
		return nil, err
	}

	// Make contact with the peer to start getting the deltas.
	var err error
	c.peer, _, err = runAtPeer(ctx, c.peer, op)
	if err != nil {
		return err
	}

	// Obtain deltas from the peer over the network.
	if err := iSt.recvAndProcessDeltas(ctx); err != nil {
		// Note, it's important to call cancel before calling Finish so that we
		// don't block waiting for the rest of the stream.
		cancel()
		// Call Finish to clean up local state even on failure.
		iSt.stream.Finish()
		return err
	}

	if err := iSt.stream.Finish(); err != nil {
		return err
	}

	vlog.VI(4).Infof("sync: getDeltas: got reply: %v", iSt.remote)

	// Process deltas locally.
	return iSt.processUpdatedObjects(ctx)
}

////////////////////////////////////////////////////////////////////////////////
// Internal helpers for initiation config.

type sgSet map[interfaces.GroupId]struct{}

// initiationConfig is the configuration information for a Database in an
// initiation round.
type initiationConfig struct {
	// Connection info of the peer to sync with. Contains mount tables that
	// this peer may have registered with. The first entry in this array is
	// the mount table where the peer was successfully reached the last
	// time. Similarly, the first entry in the neighborhood addrs is the one
	// where the peer was successfully reached the last time.
	peer connInfo

	sgIds   sgSet            // Syncgroups being requested in the initiation round.
	sgPfxs  map[string]sgSet // Syncgroup prefixes and their ids being requested in the initiation round.
	sync    *syncService
	appName string
	dbName  string
	db      interfaces.Database // handle to the Database.
	st      store.Store         // Store handle to the Database.

	// Authorizer created during the filtering of syncgroups phase. This is
	// to be used during getDeltas to authorize the peer being synced with.
	auth *namesAuthorizer
}

// newInitiatonConfig creates new initiation config. This will be shared between
// the two sync rounds in the initiation round of a Database.
func newInitiationConfig(ctx *context.T, s *syncService, name string, info sgMemberInfo, peer connInfo) (*initiationConfig, error) {
	c := &initiationConfig{}
	c.peer = peer
	c.sgIds = make(sgSet)
	for id := range info {
		c.sgIds[id] = struct{}{}
	}
	if len(c.sgIds) == 0 {
		return nil, verror.New(verror.ErrInternal, ctx, "no syncgroups found", peer.relName, name)
	}
	// Note: sgPfxs will be inited during syncgroup filtering.
	c.sync = s

	// TODO(hpucha): Would be nice to standardize on the combined "app:db"
	// name across sync (not syncbase) so we only join split/join them at
	// the boundary with the store part.
	var err error
	c.appName, c.dbName, err = splitAppDbName(ctx, name)
	if err != nil {
		return nil, err
	}

	// TODO(hpucha): nil rpc.ServerCall ok?
	c.db, err = s.getDb(ctx, nil, c.appName, c.dbName)
	if err != nil {
		return nil, err
	}

	c.st = c.db.St()

	return c, nil
}

////////////////////////////////////////////////////////////////////////////////
// Internal helpers for authorization during the two sync rounds.

// namesAuthorizer authorizes the remote peer only if it presents all the names
// in the expNames set. namesAuthorizer is created during authorizeDeltas and
// used during getDeltas.
type namesAuthorizer struct {
	expNames []string
}

func (a *namesAuthorizer) Authorize(ctx *context.T, call security.Call) error {
	names, _ := security.RemoteBlessingNames(ctx, call)
	vlog.VI(4).Infof("sync: Authorize: names %v", names)
	if len(names) == 0 {
		return verror.New(verror.ErrNoAccess, ctx)
	}
	sort.Strings(names)

	if !reflect.DeepEqual(a.expNames, names) {
		return verror.New(verror.ErrNoAccess, ctx)
	}

	vlog.VI(4).Infof("sync: Authorize: remote peer allowed")
	return nil
}

////////////////////////////////////////////////////////////////////////////////
// Internal helpers for receiving and for preliminary processing of all the log
// records over the network.

// initiationState is accumulated for a Database in each sync round in an
// initiation round.
type initiationState struct {
	// Config information.
	config *initiationConfig

	// Accumulated sync state.
	local      interfaces.Knowledge         // local generation vectors.
	remote     interfaces.Knowledge         // generation vectors from the remote peer.
	updLocal   interfaces.Knowledge         // updated local generation vectors at the end of sync round.
	updObjects map[string]*objConflictState // tracks updated objects during a log replay.
	dagGraft   *graftMap                    // DAG state that tracks conflicts and common ancestors.

	req    interfaces.DeltaReq                // GetDeltas RPC request.
	stream interfaces.SyncGetDeltasClientCall // stream handle for the GetDeltas RPC.

	// Flag to indicate if this is syncgroup metadata sync.
	sg bool

	// Transaction handle for the sync round. Used during the update
	// of objects in the Database.
	tx store.Transaction
}

// objConflictState contains the conflict state for an object that is updated
// during an initiator round.
type objConflictState struct {
	// In practice, isConflict and isAddedByCr cannot both be true.
	isAddedByCr bool
	isConflict  bool
	newHead     string
	oldHead     string
	ancestor    string
	res         *conflictResolution
	// TODO(jlodhia): Add perms object and version for the row keys for pickNew
}

// newInitiationState creates new initiation state.
func newInitiationState(ctx *context.T, c *initiationConfig, sg bool) *initiationState {
	iSt := &initiationState{}
	iSt.config = c
	iSt.updObjects = make(map[string]*objConflictState)
	iSt.dagGraft = newGraft(c.st)
	iSt.sg = sg
	return iSt
}

// prepareDataDeltaReq creates the generation vectors with local knowledge for
// the initiator to send to the responder, and creates the request to start the
// data sync.
//
// TODO(hpucha): Refactor this code with computeDelta code in sync_state.go.
func (iSt *initiationState) prepareDataDeltaReq(ctx *context.T) error {
	iSt.config.sync.thLock.Lock()
	defer iSt.config.sync.thLock.Unlock()

	// Freeze the most recent batch of local changes before fetching
	// remote changes from a peer. This frozen state is used by the
	// responder when responding to GetDeltas RPC.
	//
	// We only allow an initiator to freeze local generations (not
	// responders/watcher) in order to maintain a static baseline
	// for the duration of a sync. This addresses the following race
	// condition: If we allow responders to use newer local
	// generations while the initiator is in progress, they may beat
	// the initiator and send these new generations to remote
	// devices.  These remote devices in turn can send these
	// generations back to the initiator in progress which was
	// started with older generation information.
	if err := iSt.config.sync.checkptLocalGen(ctx, iSt.config.appName, iSt.config.dbName, nil); err != nil {
		return err
	}

	local, lgen, err := iSt.config.sync.copyDbGenInfo(ctx, iSt.config.appName, iSt.config.dbName, nil)
	if err != nil {
		return err
	}

	localPfxs := extractAndSortPrefixes(local)

	sgPfxs := make([]string, len(iSt.config.sgPfxs))
	i := 0
	for p := range iSt.config.sgPfxs {
		sgPfxs[i] = p
		i++
	}
	sort.Strings(sgPfxs)

	iSt.local = make(interfaces.Knowledge)

	if len(sgPfxs) == 0 {
		return verror.New(verror.ErrInternal, ctx, "no syncgroups for syncing")
	}

	pfx := sgPfxs[0]
	for _, p := range sgPfxs {
		if strings.HasPrefix(p, pfx) && p != pfx {
			continue
		}

		// Process this prefix as this is the start of a new set of
		// nested prefixes.
		pfx = p
		var lpStart string
		for _, lp := range localPfxs {
			if !strings.HasPrefix(lp, pfx) && !strings.HasPrefix(pfx, lp) {
				// No relationship with pfx.
				continue
			}
			if strings.HasPrefix(pfx, lp) {
				lpStart = lp
			} else {
				iSt.local[lp] = local[lp]
			}
		}
		// Deal with the starting point.
		if lpStart == "" {
			// No matching prefixes for pfx were found.
			iSt.local[pfx] = make(interfaces.GenVector)
			iSt.local[pfx][iSt.config.sync.id] = lgen
		} else {
			iSt.local[pfx] = local[lpStart]
		}
	}

	// Send request.
	req := interfaces.DataDeltaReq{
		AppName: iSt.config.appName,
		DbName:  iSt.config.dbName,
		SgIds:   iSt.config.sgIds,
		Gvs:     iSt.local,
	}

	iSt.req = interfaces.DeltaReqData{req}

	vlog.VI(4).Infof("sync: prepareDataDeltaReq: request: %v", req)

	return nil
}

// prepareSGDeltaReq creates the syncgroup generation vectors with local
// knowledge for the initiator to send to the responder, and prepares the
// request to start the syncgroup sync.
func (iSt *initiationState) prepareSGDeltaReq(ctx *context.T) error {
	iSt.config.sync.thLock.Lock()
	defer iSt.config.sync.thLock.Unlock()

	if err := iSt.config.sync.checkptLocalGen(ctx, iSt.config.appName, iSt.config.dbName, iSt.config.sgIds); err != nil {
		return err
	}

	var err error
	iSt.local, _, err = iSt.config.sync.copyDbGenInfo(ctx, iSt.config.appName, iSt.config.dbName, iSt.config.sgIds)
	if err != nil {
		return err
	}

	// Send request.
	req := interfaces.SgDeltaReq{
		AppName: iSt.config.appName,
		DbName:  iSt.config.dbName,
		Gvs:     iSt.local,
	}

	iSt.req = interfaces.DeltaReqSgs{req}

	vlog.VI(4).Infof("sync: prepareSGDeltaReq: request: %v", req)

	return nil
}

// recvAndProcessDeltas first receives the log records and generation vectors
// from the GetDeltas RPC and puts them in the Database. It also replays the
// entire log stream as the log records arrive. These records span multiple
// generations from different devices. It does not perform any conflict
// resolution during replay.  This avoids resolving conflicts that have already
// been resolved by other devices.
func (iSt *initiationState) recvAndProcessDeltas(ctx *context.T) error {
	// TODO(hpucha): This works for now, but figure out a long term solution
	// as this may be implementation dependent. It currently works because
	// the RecvStream call is stateless, and grabbing a handle to it
	// repeatedly doesn't affect what data is seen next.
	rcvr := iSt.stream.RecvStream()

	// TODO(hpucha): See if we can avoid committing the entire delta stream
	// as one batch. Currently the dependency is between the log records and
	// the batch info.
	tx := iSt.config.st.NewTransaction()
	committed := false

	defer func() {
		if !committed {
			tx.Abort()
		}
	}()

	// Track received batches (BatchId --> BatchCount mapping).
	batchMap := make(map[uint64]uint64)

	for rcvr.Advance() {
		resp := rcvr.Value()
		switch v := resp.(type) {
		case interfaces.DeltaRespGvs:
			iSt.remote = v.Value

		case interfaces.DeltaRespRec:
			// Insert log record in Database.
			// TODO(hpucha): Should we reserve more positions in a batch?
			// TODO(hpucha): Handle if syncgroup is left/destroyed while sync is in progress.
			var pos uint64
			if iSt.sg {
				pos = iSt.config.sync.reservePosInDbLog(ctx, iSt.config.appName, iSt.config.dbName, v.Value.Metadata.ObjId, 1)
			} else {
				pos = iSt.config.sync.reservePosInDbLog(ctx, iSt.config.appName, iSt.config.dbName, "", 1)
			}

			rec := &LocalLogRec{Metadata: v.Value.Metadata, Pos: pos}
			batchId := rec.Metadata.BatchId
			if batchId != NoBatchId {
				if cnt, ok := batchMap[batchId]; !ok {
					if iSt.config.sync.startBatch(ctx, tx, batchId) != batchId {
						return verror.New(verror.ErrInternal, ctx, "failed to create batch info")
					}
					batchMap[batchId] = rec.Metadata.BatchCount
				} else if cnt != rec.Metadata.BatchCount {
					return verror.New(verror.ErrInternal, ctx, "inconsistent counts for tid", batchId, cnt, rec.Metadata.BatchCount)
				}
			}

			vlog.VI(4).Infof("sync: recvAndProcessDeltas: processing rec %v", rec)
			if err := iSt.insertRecInLogAndDag(ctx, rec, batchId, tx); err != nil {
				return err
			}

			if iSt.sg {
				// Add the syncgroup value to the Database.
				if err := iSt.insertSgRecInDb(ctx, rec, v.Value.Value, tx); err != nil {
					return err
				}
			} else {
				if err := iSt.insertRecInDb(ctx, rec, v.Value.Value, tx); err != nil {
					return err
				}
				// Check for BlobRefs, and process them.
				if err := iSt.config.sync.processBlobRefs(ctx, iSt.config.peer.relName, iSt.config.sgPfxs, &rec.Metadata, v.Value.Value); err != nil {
					return err
				}
			}

			// Mark object dirty.
			iSt.updObjects[rec.Metadata.ObjId] = &objConflictState{}
		}
	}

	if err := rcvr.Err(); err != nil {
		return err
	}

	// End the started batches if any.
	for bid, cnt := range batchMap {
		if err := iSt.config.sync.endBatch(ctx, tx, bid, cnt); err != nil {
			return err
		}
	}

	// Commit this transaction. We do not retry this transaction since it
	// should not conflict with any other keys. So if it fails, it is a
	// non-retriable error.
	err := tx.Commit()
	if verror.ErrorID(err) == store.ErrConcurrentTransaction.ID {
		// Note: This might be triggered with memstore until it handles
		// transactions in a more fine-grained fashion.
		vlog.Fatalf("sync: recvAndProcessDeltas: encountered concurrent transaction")
	}
	if err == nil {
		committed = true
	}
	return err
}

// insertRecInLogAndDag adds a new log record to log and dag data structures.
func (iSt *initiationState) insertRecInLogAndDag(ctx *context.T, rec *LocalLogRec, batchId uint64, tx store.Transaction) error {
	var pfx string
	m := rec.Metadata

	if iSt.sg {
		pfx = m.ObjId
	} else {
		pfx = logDataPrefix
	}

	if err := putLogRec(ctx, tx, pfx, rec); err != nil {
		return err
	}
	logKey := logRecKey(pfx, m.Id, m.Gen)

	var err error
	switch m.RecType {
	case interfaces.NodeRec:
		err = iSt.config.sync.addNode(ctx, tx, m.ObjId, m.CurVers, logKey, m.Delete, m.Parents, m.BatchId, iSt.dagGraft)
	case interfaces.LinkRec:
		err = iSt.config.sync.addParent(ctx, tx, m.ObjId, m.CurVers, m.Parents[0], iSt.dagGraft)
	default:
		err = verror.New(verror.ErrInternal, ctx, "unknown log record type")
	}

	return err
}

// insertSgRecInDb inserts the versioned value of a syncgroup in the Database.
func (iSt *initiationState) insertSgRecInDb(ctx *context.T, rec *LocalLogRec, valbuf []byte, tx store.Transaction) error {
	m := rec.Metadata
	var sg interfaces.Syncgroup
	if err := vom.Decode(valbuf, &sg); err != nil {
		return err
	}
	return setSGDataEntryByOID(ctx, tx, m.ObjId, m.CurVers, &sg)
}

// insertRecInDb inserts the versioned value in the Database.
func (iSt *initiationState) insertRecInDb(ctx *context.T, rec *LocalLogRec, valbuf []byte, tx store.Transaction) error {
	m := rec.Metadata
	// TODO(hpucha): Hack right now. Need to change Database's handling of
	// deleted objects. Currently, the initiator needs to treat deletions
	// specially since deletions do not get a version number or a special
	// value in the Database.
	if !rec.Metadata.Delete && rec.Metadata.RecType == interfaces.NodeRec {
		return watchable.PutAtVersion(ctx, tx, []byte(m.ObjId), valbuf, []byte(m.CurVers))
	}
	return nil
}

////////////////////////////////////////////////////////////////////////////////
// Internal helpers for finishing the local processing of all the log records
// received over the network.

// processUpdatedObjects processes all the updates received by the initiator,
// one object at a time. Conflict detection and resolution is carried out after
// the entire delta of log records is replayed, instead of incrementally after
// each record/batch is replayed, to avoid repeating conflict resolution already
// performed by other peers.
//
// For each updated object, we first check if the object has any conflicts,
// resulting in three possibilities:
//
// * There is no conflict, and no updates are needed to the Database
// (isConflict=false, newHead == oldHead). All changes received convey
// information that still keeps the local head as the most recent version. This
// occurs when conflicts are resolved by picking the existing local version.
//
// * There is no conflict, but a remote version is discovered that builds on the
// local head (isConflict=false, newHead != oldHead). In this case, we generate
// a Database update to simply update the Database to the latest value.
//
// * There is a conflict and we call into the app or use a well-known policy to
// resolve the conflict, resulting in three possibilties: (a) conflict was
// resolved by picking the local version. In this case, Database need not be
// updated, but a link is added to record the choice. (b) conflict was resolved
// by picking the remote version. In this case, Database is updated with the
// remote version and a link is added as well. (c) conflict was resolved by
// generating a new Database update. In this case, Database is updated with the
// new version.
//
// We collect all the updates to the Database in a transaction. In addition, as
// part of the same transaction, we update the log and dag state suitably (move
// the head ptr of the object in the dag to the latest version, and create a new
// log record reflecting conflict resolution if any). Finally, we update the
// sync state first on storage. This transaction's commit can fail since
// preconditions on the objects may have been violated. In this case, we wait to
// get the latest versions of objects from the Database, and recheck if the object
// has any conflicts and repeat the above steps, until the transaction commits
// successfully. Upon commit, we also update the in-memory sync state of the
// Database.
func (iSt *initiationState) processUpdatedObjects(ctx *context.T) error {
	// Note that the tx handle in initiation state is cached for the scope of
	// this function only as different stages in the pipeline add to the
	// transaction.
	committed := false
	defer func() {
		if !committed {
			iSt.tx.Abort()
		}
	}()

	for {
		vlog.VI(4).Infof("sync: processUpdatedObjects: begin: %d objects updated", len(iSt.updObjects))

		iSt.tx = iSt.config.st.NewTransaction()
		watchable.SetTransactionFromSync(iSt.tx) // for echo-suppression

		if count, err := iSt.detectConflicts(ctx); err != nil {
			return err
		} else {
			vlog.VI(4).Infof("sync: processUpdatedObjects: %d conflicts detected", count)
		}

		if err := iSt.resolveConflicts(ctx); err != nil {
			return err
		}

		err := iSt.updateDbAndSyncSt(ctx)
		if err == nil {
			err = iSt.tx.Commit()
		}
		if err == nil {
			committed = true
			// Update in-memory genvectors since commit is successful.
			if err := iSt.config.sync.putDbGenInfoRemote(ctx, iSt.config.appName, iSt.config.dbName, iSt.sg, iSt.updLocal); err != nil {
				vlog.Fatalf("sync: processUpdatedObjects: putting geninfo in memory failed for app %s db %s, err %v", iSt.config.appName, iSt.config.dbName, err)
			}

			// Ignore errors.
			iSt.advertiseSyncgroups(ctx)

			vlog.VI(4).Info("sync: processUpdatedObjects: end: changes committed")
			return nil
		}
		if verror.ErrorID(err) != store.ErrConcurrentTransaction.ID {
			return err
		}

		// Either updateDbAndSyncSt() or tx.Commit() detected a
		// concurrent transaction. Retry processing the remote updates.
		//
		// TODO(hpucha): Sleeping and retrying is a temporary
		// solution. Next iteration will have coordination with watch
		// thread to intelligently retry. Hence this value is not a
		// config param.
		vlog.VI(4).Info("sync: processUpdatedObjects: retry due to local mutations")
		iSt.tx.Abort()
		time.Sleep(1 * time.Second)
		iSt.resetConflictResolutionState(ctx)
	}
}

// detectConflicts iterates through all the updated objects to detect conflicts.
func (iSt *initiationState) detectConflicts(ctx *context.T) (int, error) {
	count := 0
	for objid, confSt := range iSt.updObjects {
		// Check if object has a conflict.
		var err error
		confSt.isConflict, confSt.newHead, confSt.oldHead, confSt.ancestor, err = hasConflict(ctx, iSt.tx, objid, iSt.dagGraft)
		if err != nil {
			return 0, err
		}

		if !confSt.isConflict {
			if confSt.newHead == confSt.oldHead {
				confSt.res = &conflictResolution{ty: pickLocal}
			} else {
				confSt.res = &conflictResolution{ty: pickRemote}
			}
		} else {
			count++
		}
	}
	return count, nil
}

// resetConflictResolutionState resets the accumulated state from conflict
// resolution. This is done prior to retrying conflict resolution when the
// initiator fails to update the store with the received changes due to
// concurent mutations from the client.
func (iSt *initiationState) resetConflictResolutionState(ctx *context.T) {
	for objid, confSt := range iSt.updObjects {
		// Check if the object was added during resolution. If so,
		// remove this object from the dirty objects list before
		// retrying; else reset the conflict resolution state.
		if confSt.isAddedByCr {
			delete(iSt.updObjects, objid)
		} else {
			iSt.updObjects[objid] = &objConflictState{}
		}
	}
}

// updateDbAndSync updates the Database, and if that is successful, updates log,
// dag and genvectors data structures as needed.
func (iSt *initiationState) updateDbAndSyncSt(ctx *context.T) error {
	for objid, confSt := range iSt.updObjects {
		// If the local version is picked, no further updates to the
		// Database are needed. If the remote version is picked or if a
		// new version is created, we put it in the Database.
		if confSt.res.ty != pickLocal && !iSt.sg {

			// TODO(hpucha): Hack right now. Need to change Database's
			// handling of deleted objects.
			oldVersDeleted := true
			if confSt.oldHead != NoVersion {
				oldDagNode, err := getNode(ctx, iSt.tx, objid, confSt.oldHead)
				if err != nil {
					return err
				}
				oldVersDeleted = oldDagNode.Deleted
			}

			var newVersion string
			var newVersDeleted bool
			switch confSt.res.ty {
			case pickRemote:
				newVersion = confSt.newHead
				newDagNode, err := getNode(ctx, iSt.tx, objid, newVersion)
				if err != nil {
					return err
				}
				newVersDeleted = newDagNode.Deleted
			case createNew:
				newVersion = confSt.res.rec.Metadata.CurVers
				newVersDeleted = confSt.res.rec.Metadata.Delete
			}

			// Skip delete followed by a delete.
			if oldVersDeleted && newVersDeleted {
				continue
			}

			if !oldVersDeleted {
				// Read current version to enter it in the readset of the transaction.
				version, err := watchable.GetVersion(ctx, iSt.tx, []byte(objid))
				if err != nil {
					return err
				}
				if string(version) != confSt.oldHead {
					vlog.VI(4).Infof("sync: updateDbAndSyncSt: concurrent updates %s %s", version, confSt.oldHead)
					return store.NewErrConcurrentTransaction(ctx)
				}
			} else {
				// Ensure key doesn't exist.
				if _, err := watchable.GetVersion(ctx, iSt.tx, []byte(objid)); verror.ErrorID(err) != store.ErrUnknownKey.ID {
					return store.NewErrConcurrentTransaction(ctx)
				}
			}

			if !newVersDeleted {
				if confSt.res.ty == createNew {
					vlog.VI(4).Infof("sync: updateDbAndSyncSt: PutAtVersion %s %s", objid, newVersion)
					if err := watchable.PutAtVersion(ctx, iSt.tx, []byte(objid), confSt.res.val, []byte(newVersion)); err != nil {
						return err
					}
				}
				vlog.VI(4).Infof("sync: updateDbAndSyncSt: PutVersion %s %s", objid, newVersion)
				if err := watchable.PutVersion(ctx, iSt.tx, []byte(objid), []byte(newVersion)); err != nil {
					return err
				}
			} else {
				vlog.VI(4).Infof("sync: updateDbAndSyncSt: Deleting obj %s", objid)
				if err := iSt.tx.Delete([]byte(objid)); err != nil {
					return err
				}
			}

			// If this is a perms key, update the local store index.
			if util.IsPermsKey(objid) {
				table, row := util.ParseTableAndRowOrDie(objid)
				tb := iSt.config.db.Table(ctx, table)
				var err error
				if !newVersDeleted {
					err = tb.UpdatePrefixPermsIndexForSet(ctx, iSt.tx, row)
				} else {
					err = tb.UpdatePrefixPermsIndexForDelete(ctx, iSt.tx, row)
				}
				if err != nil {
					return err
				}
			}
		}
		// Always update sync state irrespective of local/remote/new
		// versions being picked.
		if err := iSt.updateLogAndDag(ctx, objid); err != nil {
			return err
		}
	}

	return iSt.updateSyncSt(ctx)
}

// updateLogAndDag updates the log and dag data structures.
func (iSt *initiationState) updateLogAndDag(ctx *context.T, objid string) error {
	confSt, ok := iSt.updObjects[objid]
	if !ok {
		return verror.New(verror.ErrInternal, ctx, "object state not found", objid)
	}
	var newVersion string

	if !confSt.isConflict {
		newVersion = confSt.newHead
	} else {
		// Object had a conflict. Create a log record to reflect resolution.
		var rec *LocalLogRec

		switch {
		case confSt.res.ty == pickLocal:
			// Local version was picked as the conflict resolution.
			rec = iSt.createLocalLinkLogRec(ctx, objid, confSt.oldHead, confSt.newHead)
			newVersion = confSt.oldHead
		case confSt.res.ty == pickRemote:
			// Remote version was picked as the conflict resolution.
			rec = iSt.createLocalLinkLogRec(ctx, objid, confSt.newHead, confSt.oldHead)
			newVersion = confSt.newHead
		default:
			// New version was created to resolve the conflict.
			rec = confSt.res.rec
			newVersion = confSt.res.rec.Metadata.CurVers
		}

		var pfx string
		if iSt.sg {
			pfx = objid
		} else {
			pfx = logDataPrefix
		}
		if err := putLogRec(ctx, iSt.tx, pfx, rec); err != nil {
			return err
		}

		// Add a new DAG node.
		var err error
		m := rec.Metadata
		switch m.RecType {
		case interfaces.NodeRec:
			err = iSt.config.sync.addNode(ctx, iSt.tx, objid, m.CurVers, logRecKey(pfx, m.Id, m.Gen), m.Delete, m.Parents, NoBatchId, nil)
		case interfaces.LinkRec:
			err = iSt.config.sync.addParent(ctx, iSt.tx, objid, m.CurVers, m.Parents[0], nil)
		default:
			return verror.New(verror.ErrInternal, ctx, "unknown log record type")
		}
		if err != nil {
			return err
		}
	}

	// Move the head. This should be idempotent. We may move head to the
	// local head in some cases.
	return moveHead(ctx, iSt.tx, objid, newVersion)
}

func (iSt *initiationState) createLocalLinkLogRec(ctx *context.T, objid, vers, par string) *LocalLogRec {
	vlog.VI(4).Infof("sync: createLocalLinkLogRec: obj %s vers %s par %s", objid, vers, par)

	var gen, pos uint64
	if iSt.sg {
		gen, pos = iSt.config.sync.reserveGenAndPosInDbLog(ctx, iSt.config.appName, iSt.config.dbName, objid, 1)
	} else {
		gen, pos = iSt.config.sync.reserveGenAndPosInDbLog(ctx, iSt.config.appName, iSt.config.dbName, "", 1)
	}

	rec := &LocalLogRec{
		Metadata: interfaces.LogRecMetadata{
			Id:      iSt.config.sync.id,
			Gen:     gen,
			RecType: interfaces.LinkRec,

			ObjId:      objid,
			CurVers:    vers,
			Parents:    []string{par},
			UpdTime:    time.Now().UTC(),
			BatchId:    NoBatchId,
			BatchCount: 1,
			// TODO(hpucha): What is its batchid and count?
		},
		Pos: pos,
	}
	return rec
}

// updateSyncSt updates local sync state at the end of an initiator cycle.
func (iSt *initiationState) updateSyncSt(ctx *context.T) error {
	// Get the current local sync state.
	dsInMem, err := iSt.config.sync.copyDbSyncStateInMem(ctx, iSt.config.appName, iSt.config.dbName)
	if err != nil {
		return err
	}
	// Create the state to be persisted.
	ds := &DbSyncState{
		GenVecs:   dsInMem.genvecs,
		SgGenVecs: dsInMem.sggenvecs,
	}

	genvecs := ds.GenVecs
	if iSt.sg {
		genvecs = ds.SgGenVecs
	}
	// remote can be a subset of local.
	for rpfx, respgv := range iSt.remote {
		for lpfx, lpgv := range genvecs {
			if strings.HasPrefix(lpfx, rpfx) {
				mergeGenVectors(lpgv, respgv)
			}
		}
		if _, ok := genvecs[rpfx]; !ok {
			genvecs[rpfx] = respgv
		}

		if iSt.sg {
			// Flip sync pending if needed in case of syncgroup
			// syncing. See explanation for SyncPending flag in
			// types.vdl.
			gid, err := sgID(rpfx)
			if err != nil {
				return err
			}
			state, err := getSGIdEntry(ctx, iSt.tx, gid)
			if err != nil {
				return err
			}
			if state.SyncPending {
				curgv := genvecs[rpfx]
				res := curgv.Compare(state.PendingGenVec)
				vlog.VI(4).Infof("sync: updateSyncSt: checking join pending %v, curgv %v, res %v", state.PendingGenVec, curgv, res)
				if res >= 0 {
					state.SyncPending = false
					if err := setSGIdEntry(ctx, iSt.tx, gid, state); err != nil {
						return err
					}
				}
			}
		}
	}

	iSt.updLocal = genvecs
	// Clean the genvector of any local state. Note that local state is held
	// in gen/ckPtGen in sync state struct.
	for _, gv := range iSt.updLocal {
		delete(gv, iSt.config.sync.id)
	}

	// TODO(hpucha): Add knowledge compaction.

	return putDbSyncState(ctx, iSt.tx, ds)
}

// mergeGenVectors merges responder genvector into local genvector.
func mergeGenVectors(lpgv, respgv interfaces.GenVector) {
	for devid, rgen := range respgv {
		gen, ok := lpgv[devid]
		if !ok || gen < rgen {
			lpgv[devid] = rgen
		}
	}
}

func (iSt *initiationState) advertiseSyncgroups(ctx *context.T) error {
	if !iSt.sg {
		return nil
	}

	// For all the syncgroup changes we learned, see if the latest acl makes
	// this node an admin or removes it from its admin role, and if so,
	// advertise the syncgroup or cancel the existing advertisement over the
	// neighborhood as applicable.
	for objid := range iSt.updObjects {
		gid, err := sgID(objid)
		if err != nil {
			return err
		}
		var sg *interfaces.Syncgroup
		sg, err = getSyncgroupById(ctx, iSt.config.st, gid)
		if err != nil {
			return err
		}
		if err := iSt.config.sync.advertiseSyncgroupInNeighborhood(sg); err != nil {
			return err
		}
	}

	return nil
}

////////////////////////////////////////////////////////////////////////////////
// Internal helpers to make different RPCs needed for syncing.

// remoteOp encapsulates an RPC operation run against "peer". The returned
// interface can contain a response message (e.g. TimeResp, for GetTime) or can
// be nil (e.g. for GetDeltas stream connections).
type remoteOp func(ctx *context.T, peer string) (interface{}, error)

// runAtPeer attempts to connect to the remote peer using the neighborhood
// address when specified or the mount tables obtained from all the common
// syncgroups, and runs the specified op.
func runAtPeer(ctx *context.T, peer connInfo, op remoteOp) (connInfo, interface{}, error) {
	if len(peer.mtTbls) < 1 && len(peer.addrs) < 1 {
		return peer, nil, verror.New(verror.ErrInternal, ctx, "no mount tables or endpoints found", peer)
	}

	updPeer := peer
	if peer.addrs != nil {
		for i, addr := range peer.addrs {
			absName := naming.Join(addr, util.SyncbaseSuffix)
			if resp, err := runRemoteOp(ctx, absName, op); verror.ErrorID(err) != interfaces.ErrConnFail.ID {
				updPeer.addrs = updPeer.addrs[i:]
				return updPeer, resp, err
			}
		}
		updPeer.addrs = nil
	} else {
		for i, mt := range peer.mtTbls {
			absName := naming.Join(mt, peer.relName, util.SyncbaseSuffix)
			if resp, err := runRemoteOp(ctx, absName, op); verror.ErrorID(err) != interfaces.ErrConnFail.ID {
				updPeer.mtTbls = updPeer.mtTbls[i:]
				return updPeer, resp, err
			}
		}
		updPeer.mtTbls = nil
	}

	return updPeer, nil, verror.New(interfaces.ErrConnFail, ctx, "couldn't connect to peer", updPeer.relName, updPeer.addrs)
}

func runRemoteOp(ctxIn *context.T, absName string, op remoteOp) (interface{}, error) {
	vlog.VI(4).Infof("sync: runRemoteOp: begin for %v", absName)

	ctx, cancel := context.WithCancel(ctxIn)

	// We start a timer to bound the amount of time we wait to
	// initiate a connection.
	t := time.AfterFunc(connectionTimeOut, cancel)
	resp, err := op(ctx, absName)
	t.Stop()

	if err == nil {
		vlog.VI(4).Infof("sync: runRemoteOp: end op established for %s", absName)
		return resp, nil
	}

	vlog.VI(4).Infof("sync: runRemoteOp: end for %s, err %v", absName, err)

	// TODO(hpucha): Fix error handling so that we do not assume that any error
	// that is not ErrNoAccess or ErrGetTimeFailed is a connection error. Need to
	// chat with m3b, mattr, et al to figure out how applications should
	// distinguish RPC errors from application errors in general.
	if verror.ErrorID(err) != verror.ErrNoAccess.ID && verror.ErrorID(err) != interfaces.ErrGetTimeFailed.ID {
		err = verror.New(interfaces.ErrConnFail, ctx, "couldn't connect to peer", absName)
	}

	// When the RPC is successful, cancelling the parent context will take
	// care of cancelling the child context. When the err is non-nil, we
	// cancel the context here.
	cancel()
	return nil, err
}