x/ref/services/syncbase/sync/ilog.go - roadmap.go.syncbase - Git at Google

 // 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

 // Package vsync provides veyron sync ILog utility functions.  ILog
 // (Indexed Log) provides log functionality with indexing support.
 // ILog stores log records that are locally generated or obtained over
 // the network.  Indexing is needed since sync needs to selectively
 // retrieve log records that belong to a particular device, syncroot
 // and generation during synchronization.
 //
 // When a device receives a request to send log records, it first
 // computes the missing generations between itself and the incoming
 // request on a per-syncroot basis. It then sends all the log records
 // belonging to each missing generation.  A device that receives log
 // records over the network replays all the records received from
 // another device in a single batch. Each replayed log record adds a
 // new version to the dag of the object contained in the log
 // record. At the end of replaying all the log records, conflict
 // detection and resolution is carried out for all the objects learned
 // during this iteration. Conflict detection and resolution is carried
 // out after a batch of log records are replayed, instead of
 // incrementally after each record is replayed, to avoid repeating
 // conflict resolution already performed by other devices.
 //
 // New log records are created when objects in the local store are
 // created/updated. Local log records are also replayed to keep the
 // per-object dags consistent with the local store state.
 //
 // Note that syncgroups are mainly tracked between syncd/store and the
 // client. Sync-related metadata (log records, generations and
 // generation vectors) is unaware of syncgroups, and uses syncroots
 // instead. Each syncgroup contains a root object and a unique root
 // object ID. In case of peer syncgroups, all the peer syncgroups
 // contain the same root object ID.  Sync translates any given
 // syncgroup to a syncroot rooted at this root object ID and syncs all
 // the syncroots the device is part of.  Thus, although a device may
 // be part of more than 1 peer syncgroup, at the sync level, a single
 // syncroot is synced on behalf of all the peer syncgroups.
 //
 // Implementation notes: ILog records are stored in a persistent K/V
 // database in the current implementation.  ILog db consists of 3
 // tables:
 // ** records: table consists of all the log records indexed
 // by deviceid:srid:genid:lsn referring to
 //         the device that creates the log record
 //         the root oid of the syncroot to which the log record belongs
 //         the generation on the syncroot the log record is part of
 //         the sequence number in that generation
 // Note that lsn in each generation starts from 0 and genid starts from 1.
 // ** gens: table consists of the generation metadata for each
 // generation, and is indexed by deviceid:srid:genid.
 // ** head: table consists of the log header.
 import (
 	"errors"
 	"fmt"
 	"strconv"
 	"strings"

 	"v.io/x/lib/vlog"
 )

 var (
 	errNoUpdates  = errors.New("no new local updates")
 	errInvalidLog = errors.New("invalid log db")
 )

 // curLocalGen contains metadata re. the current local generation for a SyncRoot.
 type curLocalGen struct {
 	CurGenNum GenId  // generation id for a SyncRoot's current generation.
 	CurSeqNum SeqNum // log sequence number for a SyncRoot's current generation.
 }

 // iLogHeader contains the log header metadata.
 type iLogHeader struct {
 	CurSRGens map[ObjId]*curLocalGen // local generation info per SyncRoot.
 	Curorder  uint32                 // position in log for the next generation.
 }

 // genMetadata contains the metadata for a generation.
 type genMetadata struct {
 	// All generations stored in the log are ordered wrt each
 	// other and this order needs to be preserved.
 	// Position of this generation in the log.
 	Pos uint32

 	// Number of log records in this generation still stored in the log.
 	// This count is used during garbage collection.
 	Count uint64

 	// Maximum SeqNum that was part of this generation.
 	// This is useful during garbage collection to track any unclaimed log records.
 	MaxSeqNum SeqNum
 }

 // iLog contains the metadata for the ILog db.
 type iLog struct {
 	fname string // file pathname.
 	db    *kvdb  // underlying k/v db.

 	// Key:deviceid:srid:genid:lsn Value:LogRecord. Pointer to
 	// the "records" table in the kvdb. Contains log records.
 	records *kvtable

 	// Key:deviceid:srid:genid Value:genMetadata. Pointer to the
 	// "gens" table in the kvdb. Contains generation metadata for
 	// each generation.
 	gens *kvtable

 	// Key:"Head" Value:iLogHeader. Pointer to the "header" table
 	// in the kvdb. Contains logheader.
 	header *kvtable

 	head *iLogHeader // log head cached in memory.

 	s *syncd // pointer to the sync daemon object.
 }

 // openILog opens or creates a ILog for the given filename.
 func openILog(filename string, sin *syncd) (*iLog, error) {
 	ilog := &iLog{
 		fname: filename,
 		head:  &iLogHeader{},
 		s:     sin,
 	}
 	// Open the file and create it if it does not exist.
 	// Also initialize the kvdb and its three collections.
 	db, tbls, err := kvdbOpen(filename, []string{"records", "gens", "header"})
 	if err != nil {
 		return nil, err
 	}

 	ilog.db = db
 	ilog.records = tbls[0]
 	ilog.gens = tbls[1]
 	ilog.header = tbls[2]

 	// If header already exists in db, read it back from db.
 	if ilog.hasHead() {
 		if err := ilog.getHead(); err != nil {
 			ilog.db.close() // this also closes the tables.
 			return nil, err
 		}
 	} else {
 		ilog.head.CurSRGens = make(map[ObjId]*curLocalGen)
 	}

 	return ilog, nil
 }

 // close closes the ILog and invalidate its struct.
 func (l *iLog) close() error {
 	if l.db == nil {
 		return errInvalidLog
 	}
 	// Flush the dirty data.
 	if err := l.flush(); err != nil {
 		return err
 	}

 	l.db.close() // this also closes the tables.

 	*l = iLog{} // zero out the ILog struct.
 	return nil
 }

 // flush flushes the ILog db to disk.
 func (l *iLog) flush() error {
 	if l.db == nil {
 		return errInvalidLog
 	}
 	// Set the head from memory before flushing.
 	if err := l.putHead(); err != nil {
 		return err
 	}

 	l.db.flush()

 	return nil
 }

 // initSyncRoot initializes the log header for this SyncRoot.
 func (l *iLog) initSyncRoot(srid ObjId) error {
 	if l.db == nil {
 		return errInvalidLog
 	}
 	if _, ok := l.head.CurSRGens[srid]; ok {
 		return fmt.Errorf("syncroot already exists %v", srid)
 	}
 	// First generation to be created is generation 1. A
 	// generation of 0 represents no updates on the device.
 	l.head.CurSRGens[srid] = &curLocalGen{CurGenNum: 1}
 	return nil
 }

 // delSyncRoot deletes the log header for this SyncRoot.
 func (l *iLog) delSyncRoot(srid ObjId) error {
 	if l.db == nil {
 		return errInvalidLog
 	}
 	if _, ok := l.head.CurSRGens[srid]; !ok {
 		return fmt.Errorf("syncroot doesn't exist %v", srid)
 	}

 	delete(l.head.CurSRGens, srid)
 	return nil
 }

 // putHead puts the log head into the ILog db.
 func (l *iLog) putHead() error {
 	return l.header.set("Head", l.head)
 }

 // getHead gets the log head from the ILog db.
 func (l *iLog) getHead() error {
 	if l.head == nil {
 		return errors.New("nil log header")
 	}
 	if err := l.header.get("Head", l.head); err != nil {
 		return err
 	}
 	// When we put an empty map in kvdb, we get back a "nil" map
 	// (due to VOM encoding/decoding). To keep putHead/getHead
 	// symmetrical, we add this additional initialization.
 	if l.head.CurSRGens == nil {
 		l.head.CurSRGens = make(map[ObjId]*curLocalGen)
 	}
 	return nil
 }

 // hasHead returns true if the ILog db has a log head.
 func (l *iLog) hasHead() bool {
 	return l.header.hasKey("Head")
 }

 // logRecKey creates a key for a log record.
 func logRecKey(devid DeviceId, srid ObjId, gnum GenId, lsn SeqNum) string {
 	return fmt.Sprintf("%s:%s:%d:%d", devid, srid.String(), uint64(gnum), uint64(lsn))
 }

 // strToGenId converts a string to GenId.
 func strToGenId(genIDStr string) (GenId, error) {
 	id, err := strconv.ParseUint(genIDStr, 10, 64)
 	if err != nil {
 		return 0, err
 	}
 	return GenId(id), nil
 }

 // strToObjId converts a string to ObjId.
 func strToObjId(objIDStr string) (ObjId, error) {
 	return ObjId(objIDStr), nil
 }

 // splitLogRecKey splits a : separated logrec key into its components.
 func splitLogRecKey(key string) (DeviceId, ObjId, GenId, SeqNum, error) {
 	args := strings.Split(key, ":")
 	if len(args) != 4 {
 		return "", NoObjId, 0, 0, fmt.Errorf("bad logrec key %s", key)
 	}
 	srid, err := strToObjId(args[1])
 	if err != nil {
 		return "", NoObjId, 0, 0, err
 	}
 	gnum, err := strToGenId(args[2])
 	if err != nil {
 		return "", NoObjId, 0, 0, err
 	}
 	lsn, err := strconv.ParseUint(args[3], 10, 64)
 	if err != nil {
 		return "", NoObjId, 0, 0, err
 	}
 	return DeviceId(args[0]), srid, gnum, SeqNum(lsn), nil
 }

 // putLogRec puts the log record into the ILog db.
 func (l *iLog) putLogRec(rec *LogRec) (string, error) {
 	if l.db == nil {
 		return "", errInvalidLog
 	}
 	key := logRecKey(rec.DevId, rec.SyncRootId, rec.GenNum, rec.SeqNum)
 	return key, l.records.set(key, rec)
 }

 // getLogRec gets the log record from the ILog db.
 func (l *iLog) getLogRec(devid DeviceId, srid ObjId, gnum GenId, lsn SeqNum) (*LogRec, error) {
 	if l.db == nil {
 		return nil, errInvalidLog
 	}
 	key := logRecKey(devid, srid, gnum, lsn)
 	var rec LogRec
 	if err := l.records.get(key, &rec); err != nil {
 		return nil, err
 	}
 	return &rec, nil
 }

 // hasLogRec returns true if the ILog db has a log record matching (devid, srid, gnum, lsn).
 func (l *iLog) hasLogRec(devid DeviceId, srid ObjId, gnum GenId, lsn SeqNum) bool {
 	if l.db == nil {
 		return false
 	}
 	key := logRecKey(devid, srid, gnum, lsn)
 	return l.records.hasKey(key)
 }

 // delLogRec deletes the log record matching (devid, srid, gnum, lsn) from the ILog db.
 func (l *iLog) delLogRec(devid DeviceId, srid ObjId, gnum GenId, lsn SeqNum) error {
 	if l.db == nil {
 		return errInvalidLog
 	}
 	key := logRecKey(devid, srid, gnum, lsn)
 	return l.records.del(key)
 }

 // generationKey creates a key for a generation.
 func generationKey(devid DeviceId, srid ObjId, gnum GenId) string {
 	return fmt.Sprintf("%s:%s:%d", devid, srid.String(), gnum)
 }

 // splitGenerationKey splits a : separated generation key into its components.
 func splitGenerationKey(key string) (DeviceId, ObjId, GenId, error) {
 	args := strings.Split(key, ":")
 	if len(args) != 3 {
 		return "", NoObjId, 0, fmt.Errorf("bad generation key %s", key)
 	}
 	srid, err := strToObjId(args[1])
 	if err != nil {
 		return "", NoObjId, 0, err
 	}
 	gnum, err := strToGenId(args[2])
 	if err != nil {
 		return "", NoObjId, 0, err
 	}
 	return DeviceId(args[0]), srid, gnum, nil
 }

 // putGenMetadata puts the metadata of the generation (devid, srid, gnum) into the ILog db.
 func (l *iLog) putGenMetadata(devid DeviceId, srid ObjId, gnum GenId, val *genMetadata) error {
 	key := generationKey(devid, srid, gnum)
 	return l.gens.set(key, val)
 }

 // getGenMetadata gets the metadata of the generation (devid, srid, gnum) from the ILog db.
 func (l *iLog) getGenMetadata(devid DeviceId, srid ObjId, gnum GenId) (*genMetadata, error) {
 	if l.db == nil {
 		return nil, errInvalidLog
 	}
 	key := generationKey(devid, srid, gnum)
 	var val genMetadata
 	if err := l.gens.get(key, &val); err != nil {
 		return nil, err
 	}
 	return &val, nil
 }

 // hasGenMetadata returns true if the ILog db has the generation (devid, srid, gnum).
 func (l *iLog) hasGenMetadata(devid DeviceId, srid ObjId, gnum GenId) bool {
 	key := generationKey(devid, srid, gnum)
 	return l.gens.hasKey(key)
 }

 // delGenMetadata deletes the generation (devid, srid, gnum) metadata from the ILog db.
 func (l *iLog) delGenMetadata(devid DeviceId, srid ObjId, gnum GenId) error {
 	if l.db == nil {
 		return errInvalidLog
 	}
 	key := generationKey(devid, srid, gnum)
 	return l.gens.del(key)
 }

 // getLocalGenInfo gets the local generation info for the given syncroot.
 func (l *iLog) getLocalGenInfo(srid ObjId) (*curLocalGen, error) {
 	gen, ok := l.head.CurSRGens[srid]
 	if !ok {
 		return nil, fmt.Errorf("no gen info found for srid %s", srid.String())
 	}
 	return gen, nil
 }

 // createLocalLogRec creates a new local log record of type NodeRec.
 func (l *iLog) createLocalLogRec(obj ObjId, vers Version,
 	par []Version, val *LogValue, srid ObjId) (*LogRec, error) {
 	gen, err := l.getLocalGenInfo(srid)
 	if err != nil {
 		return nil, err
 	}
 	rec := &LogRec{
 		DevId:      l.s.id,
 		SyncRootId: srid,
 		GenNum:     gen.CurGenNum,
 		SeqNum:     gen.CurSeqNum,
 		RecType:    NodeRec,

 		ObjId:   obj,
 		CurVers: vers,
 		Parents: par,
 		Value:   *val,
 	}

 	// Increment the SeqNum for the local log.
 	gen.CurSeqNum++

 	return rec, nil
 }

 // createLocalLinkLogRec creates a new local log record of type LinkRec.
 func (l *iLog) createLocalLinkLogRec(obj ObjId, vers, par Version, srid ObjId) (*LogRec, error) {
 	gen, err := l.getLocalGenInfo(srid)
 	if err != nil {
 		return nil, err
 	}
 	rec := &LogRec{
 		DevId:      l.s.id,
 		SyncRootId: srid,
 		GenNum:     gen.CurGenNum,
 		SeqNum:     gen.CurSeqNum,
 		RecType:    LinkRec,

 		ObjId:   obj,
 		CurVers: vers,
 		Parents: []Version{par},
 		Value:   LogValue{},
 	}

 	// Increment the SeqNum for the local log.
 	gen.CurSeqNum++

 	return rec, nil
 }

 // createRemoteGeneration adds a new remote generation.
 func (l *iLog) createRemoteGeneration(dev DeviceId, srid ObjId, gnum GenId, gen *genMetadata) error {
 	if l.db == nil {
 		return errInvalidLog
 	}

 	if gen.Count != uint64(gen.MaxSeqNum+1) {
 		return errors.New("mismatch in count and lsn")
 	}

 	vlog.VI(2).Infof("createRemoteGeneration:: dev %s srid %s gen %d %v", dev, srid.String(), gnum, gen)

 	gen.Pos = l.head.Curorder
 	l.head.Curorder++

 	return l.putGenMetadata(dev, srid, gnum, gen)
 }

 // createLocalGeneration creates a new local generation.
 func (l *iLog) createLocalGeneration(srid ObjId) (GenId, error) {
 	if l.db == nil {
 		return 0, errInvalidLog
 	}

 	g, err := l.getLocalGenInfo(srid)
 	if err != nil {
 		return 0, err
 	}

 	gnum := g.CurGenNum

 	// If there are no updates, there will be no new generation.
 	if g.CurSeqNum == 0 {
 		return gnum - 1, errNoUpdates
 	}

 	// Add the current generation to the db.
 	val := &genMetadata{
 		Pos:       l.head.Curorder,
 		Count:     uint64(g.CurSeqNum),
 		MaxSeqNum: g.CurSeqNum - 1,
 	}

 	err = l.putGenMetadata(l.s.id, srid, gnum, val)

 	vlog.VI(2).Infof("createLocalGeneration:: created srid %s gen %d %v", srid.String(), gnum, val)
 	// Move to the next generation irrespective of err.
 	l.head.Curorder++
 	g.CurGenNum++
 	g.CurSeqNum = 0

 	return gnum, err
 }

 // processWatchRecord processes new object versions obtained from the local store.
 func (l *iLog) processWatchRecord(objID ObjId, vers, parent Version, val *LogValue, srid ObjId) error {
 	if l.db == nil {
 		return errInvalidLog
 	}

 	vlog.VI(2).Infof("processWatchRecord:: adding for object %v %v (srid %v)", objID, vers, srid)

 	if !srid.IsValid() {
 		return errors.New("invalid syncroot id")
 	}

 	// Filter out the echo from the watcher. When syncd puts mutations into store,
 	// it hears back these mutations once again on the watch stream. We need to
 	// filter out these echoes and only process brand new watch changes.
 	if vers != NoVersion {
 		// Check if the object's vers already exists in the DAG.
 		if l.s.dag.hasNode(objID, vers) {
 			return nil
 		}

 		// When we successfully join a SyncGroup, Store
 		// creates an empty directory with object ID as the
 		// rootObjId of the SyncGroup joined.  We do not care
 		// about this version of the directory. We will start
 		// accepting local mutations on this object only after
 		// we hear about it remotely first.
 		//
 		// NOTE: Since this new directory is protected from
 		// any local updates via an Permissions until after the first
 		// sync, waiting to hear remotely first works without
 		// dropping any updates. However, we cache in the
 		// "priv" table its version in the Store so that we
 		// can correctly specify prior version when we put
 		// mutations into the store.
 		if _, err := l.s.dag.getHead(objID); err != nil && objID == srid {
 			priv := &privNode{ /*Mutation: &val.Mutation, */ SyncTime: val.SyncTime, TxId: val.TxId, TxCount: val.TxCount}
 			return l.s.dag.setPrivNode(objID, priv)
 		}
 	} else {
 		// Check if the parent version has a deleted
 		// descendant already in the DAG.
 		if l.s.dag.hasDeletedDescendant(objID, parent) {
 			return nil
 		}
 	}

 	var pars []Version
 	if parent != NoVersion {
 		pars = []Version{parent}
 	}

 	// If the current version is a deletion, generate a new version number.
 	if val.Delete {
 		if vers != NoVersion {
 			return fmt.Errorf("deleted vers is %v", vers)
 		}
 		vers = NewVersion()
 		//val.Mutation.Version = vers
 	}

 	// Create a log record from Watch's Change Record.
 	rec, err := l.createLocalLogRec(objID, vers, pars, val, srid)
 	if err != nil {
 		return err
 	}

 	// Insert the new log record into the log.
 	logKey, err := l.putLogRec(rec)
 	if err != nil {
 		return err
 	}

 	// Insert the new log record into dag.
 	if err = l.s.dag.addNode(rec.ObjId, rec.CurVers, false, val.Delete, rec.Parents, logKey, val.TxId); err != nil {
 		return err
 	}

 	// Move the head.
 	return l.s.dag.moveHead(rec.ObjId, rec.CurVers)
 }

 // dump writes to the log file information on ILog internals.
 func (l *iLog) dump() {
 	if l.db == nil {
 		return
 	}

 	vlog.VI(1).Infof("DUMP: ILog: #SR %d, cur %d", len(l.head.CurSRGens), l.head.Curorder)
 	for sr, gen := range l.head.CurSRGens {
 		vlog.VI(1).Infof("DUMP: ILog: SR %v: gen %v, lsn %v", sr, gen.CurGenNum, gen.CurSeqNum)
 	}

 	// Find for each (devid, srid) pair its lowest and highest generation numbers.
 	type genInfo struct{ min, max GenId }
 	devs := make(map[DeviceId]map[ObjId]*genInfo)

 	l.gens.keyIter(func(genKey string) {
 		devid, srid, gnum, err := splitGenerationKey(genKey)
 		if err != nil {
 			return
 		}

 		srids, ok := devs[devid]
 		if !ok {
 			srids = make(map[ObjId]*genInfo)
 			devs[devid] = srids
 		}

 		info, ok := srids[srid]
 		if ok {
 			if gnum > info.max {
 				info.max = gnum
 			}
 			if gnum < info.min {
 				info.min = gnum
 			}
 		} else {
 			srids[srid] = &genInfo{min: gnum, max: gnum}
 		}
 	})

 	// For each (devid, srid) pair dump its generation info in order from
 	// min to max generation numbers, inclusive.
 	for devid, srids := range devs {
 		for srid, info := range srids {
 			for gnum := info.min; gnum <= info.max; gnum++ {
 				meta, err := l.getGenMetadata(devid, srid, gnum)
 				if err == nil {
 					vlog.VI(1).Infof(
 						"DUMP: ILog: gen (%v, %v, %v): pos %v, count %v, maxlsn %v",
 						devid, srid, gnum, meta.Pos, meta.Count, meta.MaxSeqNum)
 				} else {
 					vlog.VI(1).Infof("DUMP: ILog: gen (%v, %v, %v): missing generation",
 						devid, srid, gnum)
 				}
 			}
 		}
 	}
 }
	// 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

	// Package vsync provides veyron sync ILog utility functions. ILog
	// (Indexed Log) provides log functionality with indexing support.
	// ILog stores log records that are locally generated or obtained over
	// the network. Indexing is needed since sync needs to selectively
	// retrieve log records that belong to a particular device, syncroot
	// and generation during synchronization.
	//
	// When a device receives a request to send log records, it first
	// computes the missing generations between itself and the incoming
	// request on a per-syncroot basis. It then sends all the log records
	// belonging to each missing generation. A device that receives log
	// records over the network replays all the records received from
	// another device in a single batch. Each replayed log record adds a
	// new version to the dag of the object contained in the log
	// record. At the end of replaying all the log records, conflict
	// detection and resolution is carried out for all the objects learned
	// during this iteration. Conflict detection and resolution is carried
	// out after a batch of log records are replayed, instead of
	// incrementally after each record is replayed, to avoid repeating
	// conflict resolution already performed by other devices.
	//
	// New log records are created when objects in the local store are
	// created/updated. Local log records are also replayed to keep the
	// per-object dags consistent with the local store state.
	//
	// Note that syncgroups are mainly tracked between syncd/store and the
	// client. Sync-related metadata (log records, generations and
	// generation vectors) is unaware of syncgroups, and uses syncroots
	// instead. Each syncgroup contains a root object and a unique root
	// object ID. In case of peer syncgroups, all the peer syncgroups
	// contain the same root object ID. Sync translates any given
	// syncgroup to a syncroot rooted at this root object ID and syncs all
	// the syncroots the device is part of. Thus, although a device may
	// be part of more than 1 peer syncgroup, at the sync level, a single
	// syncroot is synced on behalf of all the peer syncgroups.
	//
	// Implementation notes: ILog records are stored in a persistent K/V
	// database in the current implementation. ILog db consists of 3
	// tables:
	// ** records: table consists of all the log records indexed
	// by deviceid:srid:genid:lsn referring to
	// the device that creates the log record
	// the root oid of the syncroot to which the log record belongs
	// the generation on the syncroot the log record is part of
	// the sequence number in that generation
	// Note that lsn in each generation starts from 0 and genid starts from 1.
	// ** gens: table consists of the generation metadata for each
	// generation, and is indexed by deviceid:srid:genid.
	// ** head: table consists of the log header.
	import (
	"errors"
	"fmt"
	"strconv"
	"strings"

	"v.io/x/lib/vlog"
	)

	var (
	errNoUpdates = errors.New("no new local updates")
	errInvalidLog = errors.New("invalid log db")
	)

	// curLocalGen contains metadata re. the current local generation for a SyncRoot.
	type curLocalGen struct {
	CurGenNum GenId // generation id for a SyncRoot's current generation.
	CurSeqNum SeqNum // log sequence number for a SyncRoot's current generation.
	}

	// iLogHeader contains the log header metadata.
	type iLogHeader struct {
	CurSRGens map[ObjId]*curLocalGen // local generation info per SyncRoot.
	Curorder uint32 // position in log for the next generation.
	}

	// genMetadata contains the metadata for a generation.
	type genMetadata struct {
	// All generations stored in the log are ordered wrt each
	// other and this order needs to be preserved.
	// Position of this generation in the log.
	Pos uint32

	// Number of log records in this generation still stored in the log.
	// This count is used during garbage collection.
	Count uint64

	// Maximum SeqNum that was part of this generation.
	// This is useful during garbage collection to track any unclaimed log records.
	MaxSeqNum SeqNum
	}

	// iLog contains the metadata for the ILog db.
	type iLog struct {
	fname string // file pathname.
	db *kvdb // underlying k/v db.

	// Key:deviceid:srid:genid:lsn Value:LogRecord. Pointer to
	// the "records" table in the kvdb. Contains log records.
	records *kvtable

	// Key:deviceid:srid:genid Value:genMetadata. Pointer to the
	// "gens" table in the kvdb. Contains generation metadata for
	// each generation.
	gens *kvtable

	// Key:"Head" Value:iLogHeader. Pointer to the "header" table
	// in the kvdb. Contains logheader.
	header *kvtable

	head *iLogHeader // log head cached in memory.

	s *syncd // pointer to the sync daemon object.
	}

	// openILog opens or creates a ILog for the given filename.
	func openILog(filename string, sin syncd) (iLog, error) {
	ilog := &iLog{
	fname: filename,
	head: &iLogHeader{},
	s: sin,
	}
	// Open the file and create it if it does not exist.
	// Also initialize the kvdb and its three collections.
	db, tbls, err := kvdbOpen(filename, []string{"records", "gens", "header"})
	if err != nil {
	return nil, err
	}

	ilog.db = db
	ilog.records = tbls[0]
	ilog.gens = tbls[1]
	ilog.header = tbls[2]

	// If header already exists in db, read it back from db.
	if ilog.hasHead() {
	if err := ilog.getHead(); err != nil {
	ilog.db.close() // this also closes the tables.
	return nil, err
	}
	} else {
	ilog.head.CurSRGens = make(map[ObjId]*curLocalGen)
	}

	return ilog, nil
	}

	// close closes the ILog and invalidate its struct.
	func (l *iLog) close() error {
	if l.db == nil {
	return errInvalidLog
	}
	// Flush the dirty data.
	if err := l.flush(); err != nil {
	return err
	}

	l.db.close() // this also closes the tables.

	*l = iLog{} // zero out the ILog struct.
	return nil
	}

	// flush flushes the ILog db to disk.
	func (l *iLog) flush() error {
	if l.db == nil {
	return errInvalidLog
	}
	// Set the head from memory before flushing.
	if err := l.putHead(); err != nil {
	return err
	}

	l.db.flush()

	return nil
	}

	// initSyncRoot initializes the log header for this SyncRoot.
	func (l *iLog) initSyncRoot(srid ObjId) error {
	if l.db == nil {
	return errInvalidLog
	}
	if _, ok := l.head.CurSRGens[srid]; ok {
	return fmt.Errorf("syncroot already exists %v", srid)
	}
	// First generation to be created is generation 1. A
	// generation of 0 represents no updates on the device.
	l.head.CurSRGens[srid] = &curLocalGen{CurGenNum: 1}
	return nil
	}

	// delSyncRoot deletes the log header for this SyncRoot.
	func (l *iLog) delSyncRoot(srid ObjId) error {
	if l.db == nil {
	return errInvalidLog
	}
	if _, ok := l.head.CurSRGens[srid]; !ok {
	return fmt.Errorf("syncroot doesn't exist %v", srid)
	}

	delete(l.head.CurSRGens, srid)
	return nil
	}

	// putHead puts the log head into the ILog db.
	func (l *iLog) putHead() error {
	return l.header.set("Head", l.head)
	}

	// getHead gets the log head from the ILog db.
	func (l *iLog) getHead() error {
	if l.head == nil {
	return errors.New("nil log header")
	}
	if err := l.header.get("Head", l.head); err != nil {
	return err
	}
	// When we put an empty map in kvdb, we get back a "nil" map
	// (due to VOM encoding/decoding). To keep putHead/getHead
	// symmetrical, we add this additional initialization.
	if l.head.CurSRGens == nil {
	l.head.CurSRGens = make(map[ObjId]*curLocalGen)
	}
	return nil
	}

	// hasHead returns true if the ILog db has a log head.
	func (l *iLog) hasHead() bool {
	return l.header.hasKey("Head")
	}

	// logRecKey creates a key for a log record.
	func logRecKey(devid DeviceId, srid ObjId, gnum GenId, lsn SeqNum) string {
	return fmt.Sprintf("%s:%s:%d:%d", devid, srid.String(), uint64(gnum), uint64(lsn))
	}

	// strToGenId converts a string to GenId.
	func strToGenId(genIDStr string) (GenId, error) {
	id, err := strconv.ParseUint(genIDStr, 10, 64)
	if err != nil {
	return 0, err
	}
	return GenId(id), nil
	}

	// strToObjId converts a string to ObjId.
	func strToObjId(objIDStr string) (ObjId, error) {
	return ObjId(objIDStr), nil
	}

	// splitLogRecKey splits a : separated logrec key into its components.
	func splitLogRecKey(key string) (DeviceId, ObjId, GenId, SeqNum, error) {
	args := strings.Split(key, ":")
	if len(args) != 4 {
	return "", NoObjId, 0, 0, fmt.Errorf("bad logrec key %s", key)
	}
	srid, err := strToObjId(args[1])
	if err != nil {
	return "", NoObjId, 0, 0, err
	}
	gnum, err := strToGenId(args[2])
	if err != nil {
	return "", NoObjId, 0, 0, err
	}
	lsn, err := strconv.ParseUint(args[3], 10, 64)
	if err != nil {
	return "", NoObjId, 0, 0, err
	}
	return DeviceId(args[0]), srid, gnum, SeqNum(lsn), nil
	}

	// putLogRec puts the log record into the ILog db.
	func (l iLog) putLogRec(rec LogRec) (string, error) {
	if l.db == nil {
	return "", errInvalidLog
	}
	key := logRecKey(rec.DevId, rec.SyncRootId, rec.GenNum, rec.SeqNum)
	return key, l.records.set(key, rec)
	}

	// getLogRec gets the log record from the ILog db.
	func (l iLog) getLogRec(devid DeviceId, srid ObjId, gnum GenId, lsn SeqNum) (LogRec, error) {
	if l.db == nil {
	return nil, errInvalidLog
	}
	key := logRecKey(devid, srid, gnum, lsn)
	var rec LogRec
	if err := l.records.get(key, &rec); err != nil {
	return nil, err
	}
	return &rec, nil
	}

	// hasLogRec returns true if the ILog db has a log record matching (devid, srid, gnum, lsn).
	func (l *iLog) hasLogRec(devid DeviceId, srid ObjId, gnum GenId, lsn SeqNum) bool {
	if l.db == nil {
	return false
	}
	key := logRecKey(devid, srid, gnum, lsn)
	return l.records.hasKey(key)
	}

	// delLogRec deletes the log record matching (devid, srid, gnum, lsn) from the ILog db.
	func (l *iLog) delLogRec(devid DeviceId, srid ObjId, gnum GenId, lsn SeqNum) error {
	if l.db == nil {
	return errInvalidLog
	}
	key := logRecKey(devid, srid, gnum, lsn)
	return l.records.del(key)
	}

	// generationKey creates a key for a generation.
	func generationKey(devid DeviceId, srid ObjId, gnum GenId) string {
	return fmt.Sprintf("%s:%s:%d", devid, srid.String(), gnum)
	}

	// splitGenerationKey splits a : separated generation key into its components.
	func splitGenerationKey(key string) (DeviceId, ObjId, GenId, error) {
	args := strings.Split(key, ":")
	if len(args) != 3 {
	return "", NoObjId, 0, fmt.Errorf("bad generation key %s", key)
	}
	srid, err := strToObjId(args[1])
	if err != nil {
	return "", NoObjId, 0, err
	}
	gnum, err := strToGenId(args[2])
	if err != nil {
	return "", NoObjId, 0, err
	}
	return DeviceId(args[0]), srid, gnum, nil
	}

	// putGenMetadata puts the metadata of the generation (devid, srid, gnum) into the ILog db.
	func (l iLog) putGenMetadata(devid DeviceId, srid ObjId, gnum GenId, val genMetadata) error {
	key := generationKey(devid, srid, gnum)
	return l.gens.set(key, val)
	}

	// getGenMetadata gets the metadata of the generation (devid, srid, gnum) from the ILog db.
	func (l iLog) getGenMetadata(devid DeviceId, srid ObjId, gnum GenId) (genMetadata, error) {
	if l.db == nil {
	return nil, errInvalidLog
	}
	key := generationKey(devid, srid, gnum)
	var val genMetadata
	if err := l.gens.get(key, &val); err != nil {
	return nil, err
	}
	return &val, nil
	}

	// hasGenMetadata returns true if the ILog db has the generation (devid, srid, gnum).
	func (l *iLog) hasGenMetadata(devid DeviceId, srid ObjId, gnum GenId) bool {
	key := generationKey(devid, srid, gnum)
	return l.gens.hasKey(key)
	}

	// delGenMetadata deletes the generation (devid, srid, gnum) metadata from the ILog db.
	func (l *iLog) delGenMetadata(devid DeviceId, srid ObjId, gnum GenId) error {
	if l.db == nil {
	return errInvalidLog
	}
	key := generationKey(devid, srid, gnum)
	return l.gens.del(key)
	}

	// getLocalGenInfo gets the local generation info for the given syncroot.
	func (l iLog) getLocalGenInfo(srid ObjId) (curLocalGen, error) {
	gen, ok := l.head.CurSRGens[srid]
	if !ok {
	return nil, fmt.Errorf("no gen info found for srid %s", srid.String())
	}
	return gen, nil
	}

	// createLocalLogRec creates a new local log record of type NodeRec.
	func (l *iLog) createLocalLogRec(obj ObjId, vers Version,
	par []Version, val LogValue, srid ObjId) (LogRec, error) {
	gen, err := l.getLocalGenInfo(srid)
	if err != nil {
	return nil, err
	}
	rec := &LogRec{
	DevId: l.s.id,
	SyncRootId: srid,
	GenNum: gen.CurGenNum,
	SeqNum: gen.CurSeqNum,
	RecType: NodeRec,

	ObjId: obj,
	CurVers: vers,
	Parents: par,
	Value: *val,
	}

	// Increment the SeqNum for the local log.
	gen.CurSeqNum++

	return rec, nil
	}

	// createLocalLinkLogRec creates a new local log record of type LinkRec.
	func (l iLog) createLocalLinkLogRec(obj ObjId, vers, par Version, srid ObjId) (LogRec, error) {
	gen, err := l.getLocalGenInfo(srid)
	if err != nil {
	return nil, err
	}
	rec := &LogRec{
	DevId: l.s.id,
	SyncRootId: srid,
	GenNum: gen.CurGenNum,
	SeqNum: gen.CurSeqNum,
	RecType: LinkRec,

	ObjId: obj,
	CurVers: vers,
	Parents: []Version{par},
	Value: LogValue{},
	}

	// Increment the SeqNum for the local log.
	gen.CurSeqNum++

	return rec, nil
	}

	// createRemoteGeneration adds a new remote generation.
	func (l iLog) createRemoteGeneration(dev DeviceId, srid ObjId, gnum GenId, gen genMetadata) error {
	if l.db == nil {
	return errInvalidLog
	}

	if gen.Count != uint64(gen.MaxSeqNum+1) {
	return errors.New("mismatch in count and lsn")
	}

	vlog.VI(2).Infof("createRemoteGeneration:: dev %s srid %s gen %d %v", dev, srid.String(), gnum, gen)

	gen.Pos = l.head.Curorder
	l.head.Curorder++

	return l.putGenMetadata(dev, srid, gnum, gen)
	}

	// createLocalGeneration creates a new local generation.
	func (l *iLog) createLocalGeneration(srid ObjId) (GenId, error) {
	if l.db == nil {
	return 0, errInvalidLog
	}

	g, err := l.getLocalGenInfo(srid)
	if err != nil {
	return 0, err
	}

	gnum := g.CurGenNum

	// If there are no updates, there will be no new generation.
	if g.CurSeqNum == 0 {
	return gnum - 1, errNoUpdates
	}

	// Add the current generation to the db.
	val := &genMetadata{
	Pos: l.head.Curorder,
	Count: uint64(g.CurSeqNum),
	MaxSeqNum: g.CurSeqNum - 1,
	}

	err = l.putGenMetadata(l.s.id, srid, gnum, val)

	vlog.VI(2).Infof("createLocalGeneration:: created srid %s gen %d %v", srid.String(), gnum, val)
	// Move to the next generation irrespective of err.
	l.head.Curorder++
	g.CurGenNum++
	g.CurSeqNum = 0

	return gnum, err
	}

	// processWatchRecord processes new object versions obtained from the local store.
	func (l iLog) processWatchRecord(objID ObjId, vers, parent Version, val LogValue, srid ObjId) error {
	if l.db == nil {
	return errInvalidLog
	}

	vlog.VI(2).Infof("processWatchRecord:: adding for object %v %v (srid %v)", objID, vers, srid)

	if !srid.IsValid() {
	return errors.New("invalid syncroot id")
	}

	// Filter out the echo from the watcher. When syncd puts mutations into store,
	// it hears back these mutations once again on the watch stream. We need to
	// filter out these echoes and only process brand new watch changes.
	if vers != NoVersion {
	// Check if the object's vers already exists in the DAG.
	if l.s.dag.hasNode(objID, vers) {
	return nil
	}

	// When we successfully join a SyncGroup, Store
	// creates an empty directory with object ID as the
	// rootObjId of the SyncGroup joined. We do not care
	// about this version of the directory. We will start
	// accepting local mutations on this object only after
	// we hear about it remotely first.
	//
	// NOTE: Since this new directory is protected from
	// any local updates via an Permissions until after the first
	// sync, waiting to hear remotely first works without
	// dropping any updates. However, we cache in the
	// "priv" table its version in the Store so that we
	// can correctly specify prior version when we put
	// mutations into the store.
	if _, err := l.s.dag.getHead(objID); err != nil && objID == srid {
	priv := &privNode{ /Mutation: &val.Mutation, / SyncTime: val.SyncTime, TxId: val.TxId, TxCount: val.TxCount}
	return l.s.dag.setPrivNode(objID, priv)
	}
	} else {
	// Check if the parent version has a deleted
	// descendant already in the DAG.
	if l.s.dag.hasDeletedDescendant(objID, parent) {
	return nil
	}
	}

	var pars []Version
	if parent != NoVersion {
	pars = []Version{parent}
	}

	// If the current version is a deletion, generate a new version number.
	if val.Delete {
	if vers != NoVersion {
	return fmt.Errorf("deleted vers is %v", vers)
	}
	vers = NewVersion()
	//val.Mutation.Version = vers
	}

	// Create a log record from Watch's Change Record.
	rec, err := l.createLocalLogRec(objID, vers, pars, val, srid)
	if err != nil {
	return err
	}

	// Insert the new log record into the log.
	logKey, err := l.putLogRec(rec)
	if err != nil {
	return err
	}

	// Insert the new log record into dag.
	if err = l.s.dag.addNode(rec.ObjId, rec.CurVers, false, val.Delete, rec.Parents, logKey, val.TxId); err != nil {
	return err
	}

	// Move the head.
	return l.s.dag.moveHead(rec.ObjId, rec.CurVers)
	}

	// dump writes to the log file information on ILog internals.
	func (l *iLog) dump() {
	if l.db == nil {
	return
	}

	vlog.VI(1).Infof("DUMP: ILog: #SR %d, cur %d", len(l.head.CurSRGens), l.head.Curorder)
	for sr, gen := range l.head.CurSRGens {
	vlog.VI(1).Infof("DUMP: ILog: SR %v: gen %v, lsn %v", sr, gen.CurGenNum, gen.CurSeqNum)
	}

	// Find for each (devid, srid) pair its lowest and highest generation numbers.
	type genInfo struct{ min, max GenId }
	devs := make(map[DeviceId]map[ObjId]*genInfo)

	l.gens.keyIter(func(genKey string) {
	devid, srid, gnum, err := splitGenerationKey(genKey)
	if err != nil {
	return
	}

	srids, ok := devs[devid]
	if !ok {
	srids = make(map[ObjId]*genInfo)
	devs[devid] = srids
	}

	info, ok := srids[srid]
	if ok {
	if gnum > info.max {
	info.max = gnum
	}
	if gnum < info.min {
	info.min = gnum
	}
	} else {
	srids[srid] = &genInfo{min: gnum, max: gnum}
	}
	})

	// For each (devid, srid) pair dump its generation info in order from
	// min to max generation numbers, inclusive.
	for devid, srids := range devs {
	for srid, info := range srids {
	for gnum := info.min; gnum <= info.max; gnum++ {
	meta, err := l.getGenMetadata(devid, srid, gnum)
	if err == nil {
	vlog.VI(1).Infof(
	"DUMP: ILog: gen (%v, %v, %v): pos %v, count %v, maxlsn %v",
	devid, srid, gnum, meta.Pos, meta.Count, meta.MaxSeqNum)
	} else {
	vlog.VI(1).Infof("DUMP: ILog: gen (%v, %v, %v): missing generation",
	devid, srid, gnum)
	}
	}
	}
	}
	}