services/syncbase/vsync/replay_test.go - release.go.x.ref - 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

 // Used to ease the setup of sync test scenarios.
 // Parses a sync command file and returns a vector of commands to execute.
 // dagReplayCommands() executes the parsed commands at the DAG API level.

 import (
 	"bufio"
 	"container/list"
 	"fmt"
 	"os"
 	"strconv"
 	"strings"
 	"testing"
 	"time"

 	"v.io/v23/context"
 	"v.io/v23/verror"
 	"v.io/v23/vom"
 	"v.io/x/ref/services/syncbase/common"
 	"v.io/x/ref/services/syncbase/server/interfaces"
 	"v.io/x/ref/services/syncbase/store/watchable"
 )

 const (
 	addLocal = iota
 	addRemote
 	linkLocal
 	linkRemote
 	genvec
 )

 var (
 	constTime = time.Date(2009, time.November, 10, 23, 0, 0, 0, time.UTC)
 )

 type syncCommand struct {
 	cmd        int
 	oid        string
 	version    string
 	parents    []string
 	logrec     string
 	deleted    bool
 	batchId    uint64
 	batchCount uint64
 	genVecs    interfaces.Knowledge
 }

 // parseSyncCommands parses a sync test file and returns its commands.
 func parseSyncCommands(file string) ([]syncCommand, error) {
 	cmds := []syncCommand{}

 	sf, err := os.Open("testdata/" + file)
 	if err != nil {
 		return nil, err
 	}
 	defer sf.Close()

 	scanner := bufio.NewScanner(sf)
 	lineno := 0
 	for scanner.Scan() {
 		lineno++
 		line := strings.TrimSpace(scanner.Text())
 		if line == "" || line[0] == '#' {
 			continue
 		}

 		// The current line encodes a command, i.e. it is not a comment line.
 		// Use strconv.Unquote to convert \xfe to the desired byte (for example).
 		// Note, we must wrap the original line in quotes before passing it to
 		// strconv.Unquote since strconv.Unquote expects the input string to look
 		// like a Go string literal (quoted).
 		qline := "\"" + line + "\""
 		if line, err = strconv.Unquote(qline); err != nil {
 			return nil, fmt.Errorf("%s: %s", err, qline)
 		}

 		args := strings.Split(line, "|")
 		nargs := len(args)

 		switch args[0] {
 		case "addl", "addr":
 			expNargs := 9
 			if nargs != expNargs {
 				return nil, fmt.Errorf("%s:%d: need %d args instead of %d",
 					file, lineno, expNargs, nargs)
 			}
 			var parents []string
 			for i := 3; i <= 4; i++ {
 				if args[i] != "" {
 					parents = append(parents, args[i])
 				}
 			}

 			batchId, err := strconv.ParseUint(args[6], 10, 64)
 			if err != nil {
 				return nil, fmt.Errorf("%s:%d: invalid batchId: %s", file, lineno, args[6])
 			}
 			batchCount, err := strconv.ParseUint(args[7], 10, 64)
 			if err != nil {
 				return nil, fmt.Errorf("%s:%d: invalid batch count: %s", file, lineno, args[7])
 			}
 			del, err := strconv.ParseBool(args[8])
 			if err != nil {
 				return nil, fmt.Errorf("%s:%d: invalid deleted bit: %s", file, lineno, args[8])
 			}
 			cmd := syncCommand{
 				oid:        args[1],
 				version:    args[2],
 				parents:    parents,
 				logrec:     args[5],
 				batchId:    batchId,
 				batchCount: batchCount,
 				deleted:    del,
 			}
 			if args[0] == "addl" {
 				cmd.cmd = addLocal
 			} else {
 				cmd.cmd = addRemote
 			}
 			cmds = append(cmds, cmd)

 		case "linkl", "linkr":
 			expNargs := 6
 			if nargs != expNargs {
 				return nil, fmt.Errorf("%s:%d: need %d args instead of %d",
 					file, lineno, expNargs, nargs)
 			}

 			if args[3] == "" {
 				return nil, fmt.Errorf("%s:%d: parent version not specified", file, lineno)
 			}
 			if args[4] != "" {
 				return nil, fmt.Errorf("%s:%d: cannot specify a 2nd parent: %s",
 					file, lineno, args[4])
 			}

 			cmd := syncCommand{
 				oid:     args[1],
 				version: args[2],
 				parents: []string{args[3]},
 				logrec:  args[5],
 			}
 			if args[0] == "linkl" {
 				cmd.cmd = linkLocal
 			} else {
 				cmd.cmd = linkRemote
 			}
 			cmds = append(cmds, cmd)

 		case "genvec":
 			cmd := syncCommand{
 				cmd:     genvec,
 				genVecs: make(interfaces.Knowledge),
 			}
 			for i := 1; i < len(args); i = i + 2 {
 				pfx := args[i]
 				genVec := make(interfaces.GenVector)
 				for _, elem := range strings.Split(args[i+1], ",") {
 					kv := strings.Split(elem, ":")
 					if len(kv) != 2 {
 						return nil, fmt.Errorf("%s:%d: invalid gen vector key/val: %s", file, lineno, elem)
 					}
 					dev, err := strconv.ParseUint(kv[0], 10, 64)
 					if err != nil {
 						return nil, fmt.Errorf("%s:%d: invalid devid: %s", file, lineno, args[i+1])
 					}
 					gen, err := strconv.ParseUint(kv[1], 10, 64)
 					if err != nil {
 						return nil, fmt.Errorf("%s:%d: invalid gen: %s", file, lineno, args[i+1])
 					}
 					genVec[dev] = gen
 				}
 				cmd.genVecs[pfx] = genVec
 			}
 			cmds = append(cmds, cmd)

 		default:
 			return nil, fmt.Errorf("%s:%d: invalid operation: %s", file, lineno, args[0])
 		}
 	}

 	err = scanner.Err()
 	return cmds, err
 }

 // dagReplayCommands parses a sync test file and replays its commands, updating
 // the DAG structures associated with the sync service.
 func (s *syncService) dagReplayCommands(ctx *context.T, syncfile string) (*graftMap, error) {
 	cmds, err := parseSyncCommands(syncfile)
 	if err != nil {
 		return nil, err
 	}

 	st := s.sv.St()
 	graft := newGraft(st)

 	for _, cmd := range cmds {
 		tx := st.NewTransaction()

 		switch cmd.cmd {
 		case addLocal:
 			err = s.addNode(ctx, tx, cmd.oid, cmd.version, cmd.logrec,
 				cmd.deleted, cmd.parents, NoBatchId, nil)
 			if err != nil {
 				return nil, fmt.Errorf("cannot add local node %s:%s: %v",
 					cmd.oid, cmd.version, err)
 			}

 			if err = moveHead(ctx, tx, cmd.oid, cmd.version); err != nil {
 				return nil, fmt.Errorf("cannot move head to %s:%s: %v",
 					cmd.oid, cmd.version, err)
 			}

 		case addRemote:
 			err = s.addNode(ctx, tx, cmd.oid, cmd.version, cmd.logrec,
 				cmd.deleted, cmd.parents, NoBatchId, graft)
 			if err != nil {
 				return nil, fmt.Errorf("cannot add remote node %s:%s: %v",
 					cmd.oid, cmd.version, err)
 			}

 		case linkLocal:
 			if err = s.addParent(ctx, tx, cmd.oid, cmd.version, cmd.parents[0], NoBatchId, nil); err != nil {
 				return nil, fmt.Errorf("cannot add local parent %s to node %s:%s: %v",
 					cmd.parents[0], cmd.oid, cmd.version, err)
 			}

 		case linkRemote:
 			if err = s.addParent(ctx, tx, cmd.oid, cmd.version, cmd.parents[0], NoBatchId, graft); err != nil {
 				return nil, fmt.Errorf("cannot add remote parent %s to node %s:%s: %v",
 					cmd.parents[0], cmd.oid, cmd.version, err)
 			}
 		}

 		tx.Commit()
 	}

 	return graft, nil
 }

 // dummyStream emulates stream of log records received from RPC.
 type dummyStream struct {
 	l     *list.List
 	entry interfaces.DeltaResp
 }

 func newStream() *dummyStream {
 	ds := &dummyStream{
 		l: list.New(),
 	}
 	return ds
 }

 func (ds *dummyStream) add(entry interfaces.DeltaResp) {
 	ds.l.PushBack(entry)
 }

 func (ds *dummyStream) Advance() bool {
 	if ds.l.Len() > 0 {
 		ds.entry = ds.l.Remove(ds.l.Front()).(interfaces.DeltaResp)
 		return true
 	}
 	return false
 }

 func (ds *dummyStream) Value() interfaces.DeltaResp {
 	return ds.entry
 }

 func (ds *dummyStream) RecvStream() interface {
 	Advance() bool
 	Value() interfaces.DeltaResp
 	Err() error
 } {
 	return ds
 }

 func (*dummyStream) Err() error { return nil }

 func (ds *dummyStream) Finish() (interfaces.DeltaFinalResp, error) {
 	return interfaces.DeltaFinalResp{}, nil
 }

 func (ds *dummyStream) Cancel() {
 }

 // replayLocalCommands replays local log records parsed from the input file.
 func replayLocalCommands(t *testing.T, s *mockService, syncfile string) {
 	cmds, err := parseSyncCommands(syncfile)
 	if err != nil {
 		t.Fatalf("parseSyncCommands failed with err %v", err)
 	}

 	tx := createDatabase(t, s).St().NewWatchableTransaction()
 	var pos uint64
 	for _, cmd := range cmds {
 		switch cmd.cmd {
 		case addLocal:
 			rec := &LocalLogRec{
 				Metadata: createMetadata(t, interfaces.NodeRec, cmd),
 				Pos:      pos,
 			}
 			err = s.sync.processLocalLogRec(nil, tx, rec)
 			if err != nil {
 				t.Fatalf("processLocalLogRec failed with err %v", err)
 			}

 			// Add to Store.
 			err = watchable.PutVersion(nil, tx, []byte(rec.Metadata.ObjId), []byte(rec.Metadata.CurVers))
 			if err != nil {
 				t.Fatalf("PutVersion failed with err %v", err)
 			}
 			err = watchable.PutAtVersion(nil, tx, []byte(rec.Metadata.ObjId), []byte("abc"), []byte(rec.Metadata.CurVers))
 			if err != nil {
 				t.Fatalf("PutAtVersion failed with err %v", err)
 			}

 		default:
 			t.Fatalf("replayLocalCommands failed with unknown command %v", cmd)
 		}
 		pos++
 	}
 	if err := tx.Commit(); err != nil {
 		t.Fatalf("cannot commit local log records %s, err %v", syncfile, err)
 	}
 }

 // createReplayStream creates a dummy stream of log records parsed from the input file.
 func createReplayStream(t *testing.T, syncfile string) *dummyStream {
 	cmds, err := parseSyncCommands(syncfile)
 	if err != nil {
 		t.Fatalf("parseSyncCommands failed with err %v", err)
 	}

 	stream := newStream()
 	for _, cmd := range cmds {
 		var ty byte
 		switch cmd.cmd {
 		case genvec:
 			gvs := interfaces.DeltaRespGvs{cmd.genVecs}
 			stream.add(gvs)
 			continue
 		case addRemote:
 			ty = interfaces.NodeRec
 		case linkRemote:
 			ty = interfaces.LinkRec
 		default:
 			t.Fatalf("createReplayStream unknown command %v", cmd)
 		}

 		var val string = "abc"
 		rawValue, err := vom.RawBytesFromValue(val)
 		if err != nil {
 			t.Fatalf("createReplayStream encode failed, err %v", err)
 		}

 		rec := interfaces.DeltaRespRec{interfaces.LogRec{
 			Metadata: createMetadata(t, ty, cmd),
 			Value:    rawValue,
 		}}

 		stream.add(rec)
 	}
 	return stream
 }

 func createMetadata(t *testing.T, ty byte, cmd syncCommand) interfaces.LogRecMetadata {
 	_, id, gen, err := splitLogRecKey(nil, cmd.logrec)
 	if err != nil {
 		t.Fatalf("createReplayStream splitLogRecKey failed, key %s, err %v", cmd.logrec, gen)
 	}
 	m := interfaces.LogRecMetadata{
 		Id:         id,
 		Gen:        gen,
 		RecType:    ty,
 		ObjId:      common.JoinKeyParts(common.RowPrefix, cmd.oid),
 		CurVers:    cmd.version,
 		Parents:    cmd.parents,
 		UpdTime:    constTime,
 		Delete:     cmd.deleted,
 		BatchId:    cmd.batchId,
 		BatchCount: cmd.batchCount,
 	}
 	return m
 }

 // splitLogRecKey is the inverse of logRecKey and returns the prefix, device id
 // and generation number.
 func splitLogRecKey(ctx *context.T, key string) (string, uint64, uint64, error) {
 	parts := common.SplitKeyParts(key)
 	verr := verror.New(verror.ErrInternal, ctx, "invalid logreckey", key)
 	if len(parts) != 5 && len(parts) != 7 {
 		return "", 0, 0, verr
 	}
 	if common.JoinKeyParts(parts[:2]...) != logPrefix {
 		return "", 0, 0, verr
 	}

 	var idStr, genStr, prefix string
 	if len(parts) == 5 {
 		if parts[2] != logDataPrefix {
 			return "", 0, 0, verr
 		}
 		prefix, idStr, genStr = parts[2], parts[3], parts[4]
 	} else { // len(parts) == 7
 		if _, err := strconv.ParseUint(parts[4], 10, 64); err != nil { // GroupId
 			return "", 0, 0, verr
 		}
 		prefix, idStr, genStr = common.JoinKeyParts(parts[2:5]...), parts[5], parts[6]
 	}
 	id, err := strconv.ParseUint(idStr, 10, 64)
 	if err != nil {
 		return "", 0, 0, verr
 	}
 	gen, err := strconv.ParseUint(genStr, 16, 64)
 	if err != nil {
 		return "", 0, 0, verr
 	}
 	return prefix, id, gen, nil
 }

 func TestSplitLogRecKey(t *testing.T) {
 	invalid := []string{
 		"y\xfe100\xfebb",
 		"l\xfe100\xfebb",
 		"y\xfel\xfed\xfe100\xfexx",
 		"y\xfel\xfed\xfeaa\xfe100",
 		"y\xfel\xfex\xfe100\xfe100",
 		"y\xfel\xfed\xfe100",
 		"y\xfel\xfed\xfe100\xfe100\xfebb",
 		"y\xfel\xfey\xfes\xfexx\xfe100\xfe100",
 	}

 	for _, k := range invalid {
 		if _, _, _, err := splitLogRecKey(nil, k); err == nil {
 			t.Fatalf("splitting log rec key didn't fail %q", k)
 		}
 	}

 	valid := []struct {
 		pfx string
 		id  uint64
 		gen uint64
 	}{
 		{logDataPrefix, 10, 20},
 		{"y\xfes\xfe2500", 190, 540},
 		{"y\xfes\xfe4200", 9999, 999999},
 	}

 	for _, v := range valid {
 		gotPfx, gotId, gotGen, err := splitLogRecKey(nil, logRecKey(v.pfx, v.id, v.gen))
 		if gotPfx != v.pfx || gotId != v.id || gotGen != v.gen || err != nil {
 			t.Fatalf("failed key conversion: pfx got %v want %v, id got %v want %v, gen got %v want %v, err %v", gotPfx, v.pfx, gotId, v.id, gotGen, v.gen, err)
 		}
 	}
 }
	// 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

	// Used to ease the setup of sync test scenarios.
	// Parses a sync command file and returns a vector of commands to execute.
	// dagReplayCommands() executes the parsed commands at the DAG API level.

	import (
	"bufio"
	"container/list"
	"fmt"
	"os"
	"strconv"
	"strings"
	"testing"
	"time"

	"v.io/v23/context"
	"v.io/v23/verror"
	"v.io/v23/vom"
	"v.io/x/ref/services/syncbase/common"
	"v.io/x/ref/services/syncbase/server/interfaces"
	"v.io/x/ref/services/syncbase/store/watchable"
	)

	const (
	addLocal = iota
	addRemote
	linkLocal
	linkRemote
	genvec
	)

	var (
	constTime = time.Date(2009, time.November, 10, 23, 0, 0, 0, time.UTC)
	)

	type syncCommand struct {
	cmd int
	oid string
	version string
	parents []string
	logrec string
	deleted bool
	batchId uint64
	batchCount uint64
	genVecs interfaces.Knowledge
	}

	// parseSyncCommands parses a sync test file and returns its commands.
	func parseSyncCommands(file string) ([]syncCommand, error) {
	cmds := []syncCommand{}

	sf, err := os.Open("testdata/" + file)
	if err != nil {
	return nil, err
	}
	defer sf.Close()

	scanner := bufio.NewScanner(sf)
	lineno := 0
	for scanner.Scan() {
	lineno++
	line := strings.TrimSpace(scanner.Text())
	if line == "" \|\| line[0] == '#' {
	continue
	}

	// The current line encodes a command, i.e. it is not a comment line.
	// Use strconv.Unquote to convert \xfe to the desired byte (for example).
	// Note, we must wrap the original line in quotes before passing it to
	// strconv.Unquote since strconv.Unquote expects the input string to look
	// like a Go string literal (quoted).
	qline := "\"" + line + "\""
	if line, err = strconv.Unquote(qline); err != nil {
	return nil, fmt.Errorf("%s: %s", err, qline)
	}

	args := strings.Split(line, "\|")
	nargs := len(args)

	switch args[0] {
	case "addl", "addr":
	expNargs := 9
	if nargs != expNargs {
	return nil, fmt.Errorf("%s:%d: need %d args instead of %d",
	file, lineno, expNargs, nargs)
	}
	var parents []string
	for i := 3; i <= 4; i++ {
	if args[i] != "" {
	parents = append(parents, args[i])
	}
	}

	batchId, err := strconv.ParseUint(args[6], 10, 64)
	if err != nil {
	return nil, fmt.Errorf("%s:%d: invalid batchId: %s", file, lineno, args[6])
	}
	batchCount, err := strconv.ParseUint(args[7], 10, 64)
	if err != nil {
	return nil, fmt.Errorf("%s:%d: invalid batch count: %s", file, lineno, args[7])
	}
	del, err := strconv.ParseBool(args[8])
	if err != nil {
	return nil, fmt.Errorf("%s:%d: invalid deleted bit: %s", file, lineno, args[8])
	}
	cmd := syncCommand{
	oid: args[1],
	version: args[2],
	parents: parents,
	logrec: args[5],
	batchId: batchId,
	batchCount: batchCount,
	deleted: del,
	}
	if args[0] == "addl" {
	cmd.cmd = addLocal
	} else {
	cmd.cmd = addRemote
	}
	cmds = append(cmds, cmd)

	case "linkl", "linkr":
	expNargs := 6
	if nargs != expNargs {
	return nil, fmt.Errorf("%s:%d: need %d args instead of %d",
	file, lineno, expNargs, nargs)
	}

	if args[3] == "" {
	return nil, fmt.Errorf("%s:%d: parent version not specified", file, lineno)
	}
	if args[4] != "" {
	return nil, fmt.Errorf("%s:%d: cannot specify a 2nd parent: %s",
	file, lineno, args[4])
	}

	cmd := syncCommand{
	oid: args[1],
	version: args[2],
	parents: []string{args[3]},
	logrec: args[5],
	}
	if args[0] == "linkl" {
	cmd.cmd = linkLocal
	} else {
	cmd.cmd = linkRemote
	}
	cmds = append(cmds, cmd)

	case "genvec":
	cmd := syncCommand{
	cmd: genvec,
	genVecs: make(interfaces.Knowledge),
	}
	for i := 1; i < len(args); i = i + 2 {
	pfx := args[i]
	genVec := make(interfaces.GenVector)
	for _, elem := range strings.Split(args[i+1], ",") {
	kv := strings.Split(elem, ":")
	if len(kv) != 2 {
	return nil, fmt.Errorf("%s:%d: invalid gen vector key/val: %s", file, lineno, elem)
	}
	dev, err := strconv.ParseUint(kv[0], 10, 64)
	if err != nil {
	return nil, fmt.Errorf("%s:%d: invalid devid: %s", file, lineno, args[i+1])
	}
	gen, err := strconv.ParseUint(kv[1], 10, 64)
	if err != nil {
	return nil, fmt.Errorf("%s:%d: invalid gen: %s", file, lineno, args[i+1])
	}
	genVec[dev] = gen
	}
	cmd.genVecs[pfx] = genVec
	}
	cmds = append(cmds, cmd)

	default:
	return nil, fmt.Errorf("%s:%d: invalid operation: %s", file, lineno, args[0])
	}
	}

	err = scanner.Err()
	return cmds, err
	}

	// dagReplayCommands parses a sync test file and replays its commands, updating
	// the DAG structures associated with the sync service.
	func (s syncService) dagReplayCommands(ctx context.T, syncfile string) (*graftMap, error) {
	cmds, err := parseSyncCommands(syncfile)
	if err != nil {
	return nil, err
	}

	st := s.sv.St()
	graft := newGraft(st)

	for _, cmd := range cmds {
	tx := st.NewTransaction()

	switch cmd.cmd {
	case addLocal:
	err = s.addNode(ctx, tx, cmd.oid, cmd.version, cmd.logrec,
	cmd.deleted, cmd.parents, NoBatchId, nil)
	if err != nil {
	return nil, fmt.Errorf("cannot add local node %s:%s: %v",
	cmd.oid, cmd.version, err)
	}

	if err = moveHead(ctx, tx, cmd.oid, cmd.version); err != nil {
	return nil, fmt.Errorf("cannot move head to %s:%s: %v",
	cmd.oid, cmd.version, err)
	}

	case addRemote:
	err = s.addNode(ctx, tx, cmd.oid, cmd.version, cmd.logrec,
	cmd.deleted, cmd.parents, NoBatchId, graft)
	if err != nil {
	return nil, fmt.Errorf("cannot add remote node %s:%s: %v",
	cmd.oid, cmd.version, err)
	}

	case linkLocal:
	if err = s.addParent(ctx, tx, cmd.oid, cmd.version, cmd.parents[0], NoBatchId, nil); err != nil {
	return nil, fmt.Errorf("cannot add local parent %s to node %s:%s: %v",
	cmd.parents[0], cmd.oid, cmd.version, err)
	}

	case linkRemote:
	if err = s.addParent(ctx, tx, cmd.oid, cmd.version, cmd.parents[0], NoBatchId, graft); err != nil {
	return nil, fmt.Errorf("cannot add remote parent %s to node %s:%s: %v",
	cmd.parents[0], cmd.oid, cmd.version, err)
	}
	}

	tx.Commit()
	}

	return graft, nil
	}

	// dummyStream emulates stream of log records received from RPC.
	type dummyStream struct {
	l *list.List
	entry interfaces.DeltaResp
	}

	func newStream() *dummyStream {
	ds := &dummyStream{
	l: list.New(),
	}
	return ds
	}

	func (ds *dummyStream) add(entry interfaces.DeltaResp) {
	ds.l.PushBack(entry)
	}

	func (ds *dummyStream) Advance() bool {
	if ds.l.Len() > 0 {
	ds.entry = ds.l.Remove(ds.l.Front()).(interfaces.DeltaResp)
	return true
	}
	return false
	}

	func (ds *dummyStream) Value() interfaces.DeltaResp {
	return ds.entry
	}

	func (ds *dummyStream) RecvStream() interface {
	Advance() bool
	Value() interfaces.DeltaResp
	Err() error
	} {
	return ds
	}

	func (*dummyStream) Err() error { return nil }

	func (ds *dummyStream) Finish() (interfaces.DeltaFinalResp, error) {
	return interfaces.DeltaFinalResp{}, nil
	}

	func (ds *dummyStream) Cancel() {
	}

	// replayLocalCommands replays local log records parsed from the input file.
	func replayLocalCommands(t testing.T, s mockService, syncfile string) {
	cmds, err := parseSyncCommands(syncfile)
	if err != nil {
	t.Fatalf("parseSyncCommands failed with err %v", err)
	}

	tx := createDatabase(t, s).St().NewWatchableTransaction()
	var pos uint64
	for _, cmd := range cmds {
	switch cmd.cmd {
	case addLocal:
	rec := &LocalLogRec{
	Metadata: createMetadata(t, interfaces.NodeRec, cmd),
	Pos: pos,
	}
	err = s.sync.processLocalLogRec(nil, tx, rec)
	if err != nil {
	t.Fatalf("processLocalLogRec failed with err %v", err)
	}

	// Add to Store.
	err = watchable.PutVersion(nil, tx, []byte(rec.Metadata.ObjId), []byte(rec.Metadata.CurVers))
	if err != nil {
	t.Fatalf("PutVersion failed with err %v", err)
	}
	err = watchable.PutAtVersion(nil, tx, []byte(rec.Metadata.ObjId), []byte("abc"), []byte(rec.Metadata.CurVers))
	if err != nil {
	t.Fatalf("PutAtVersion failed with err %v", err)
	}

	default:
	t.Fatalf("replayLocalCommands failed with unknown command %v", cmd)
	}
	pos++
	}
	if err := tx.Commit(); err != nil {
	t.Fatalf("cannot commit local log records %s, err %v", syncfile, err)
	}
	}

	// createReplayStream creates a dummy stream of log records parsed from the input file.
	func createReplayStream(t testing.T, syncfile string) dummyStream {
	cmds, err := parseSyncCommands(syncfile)
	if err != nil {
	t.Fatalf("parseSyncCommands failed with err %v", err)
	}

	stream := newStream()
	for _, cmd := range cmds {
	var ty byte
	switch cmd.cmd {
	case genvec:
	gvs := interfaces.DeltaRespGvs{cmd.genVecs}
	stream.add(gvs)
	continue
	case addRemote:
	ty = interfaces.NodeRec
	case linkRemote:
	ty = interfaces.LinkRec
	default:
	t.Fatalf("createReplayStream unknown command %v", cmd)
	}

	var val string = "abc"
	rawValue, err := vom.RawBytesFromValue(val)
	if err != nil {
	t.Fatalf("createReplayStream encode failed, err %v", err)
	}

	rec := interfaces.DeltaRespRec{interfaces.LogRec{
	Metadata: createMetadata(t, ty, cmd),
	Value: rawValue,
	}}

	stream.add(rec)
	}
	return stream
	}

	func createMetadata(t *testing.T, ty byte, cmd syncCommand) interfaces.LogRecMetadata {
	_, id, gen, err := splitLogRecKey(nil, cmd.logrec)
	if err != nil {
	t.Fatalf("createReplayStream splitLogRecKey failed, key %s, err %v", cmd.logrec, gen)
	}
	m := interfaces.LogRecMetadata{
	Id: id,
	Gen: gen,
	RecType: ty,
	ObjId: common.JoinKeyParts(common.RowPrefix, cmd.oid),
	CurVers: cmd.version,
	Parents: cmd.parents,
	UpdTime: constTime,
	Delete: cmd.deleted,
	BatchId: cmd.batchId,
	BatchCount: cmd.batchCount,
	}
	return m
	}

	// splitLogRecKey is the inverse of logRecKey and returns the prefix, device id
	// and generation number.
	func splitLogRecKey(ctx *context.T, key string) (string, uint64, uint64, error) {
	parts := common.SplitKeyParts(key)
	verr := verror.New(verror.ErrInternal, ctx, "invalid logreckey", key)
	if len(parts) != 5 && len(parts) != 7 {
	return "", 0, 0, verr
	}
	if common.JoinKeyParts(parts[:2]...) != logPrefix {
	return "", 0, 0, verr
	}

	var idStr, genStr, prefix string
	if len(parts) == 5 {
	if parts[2] != logDataPrefix {
	return "", 0, 0, verr
	}
	prefix, idStr, genStr = parts[2], parts[3], parts[4]
	} else { // len(parts) == 7
	if _, err := strconv.ParseUint(parts[4], 10, 64); err != nil { // GroupId
	return "", 0, 0, verr
	}
	prefix, idStr, genStr = common.JoinKeyParts(parts[2:5]...), parts[5], parts[6]
	}
	id, err := strconv.ParseUint(idStr, 10, 64)
	if err != nil {
	return "", 0, 0, verr
	}
	gen, err := strconv.ParseUint(genStr, 16, 64)
	if err != nil {
	return "", 0, 0, verr
	}
	return prefix, id, gen, nil
	}

	func TestSplitLogRecKey(t *testing.T) {
	invalid := []string{
	"y\xfe100\xfebb",
	"l\xfe100\xfebb",
	"y\xfel\xfed\xfe100\xfexx",
	"y\xfel\xfed\xfeaa\xfe100",
	"y\xfel\xfex\xfe100\xfe100",
	"y\xfel\xfed\xfe100",
	"y\xfel\xfed\xfe100\xfe100\xfebb",
	"y\xfel\xfey\xfes\xfexx\xfe100\xfe100",
	}

	for _, k := range invalid {
	if _, _, _, err := splitLogRecKey(nil, k); err == nil {
	t.Fatalf("splitting log rec key didn't fail %q", k)
	}
	}

	valid := []struct {
	pfx string
	id uint64
	gen uint64
	}{
	{logDataPrefix, 10, 20},
	{"y\xfes\xfe2500", 190, 540},
	{"y\xfes\xfe4200", 9999, 999999},
	}

	for _, v := range valid {
	gotPfx, gotId, gotGen, err := splitLogRecKey(nil, logRecKey(v.pfx, v.id, v.gen))
	if gotPfx != v.pfx \|\| gotId != v.id \|\| gotGen != v.gen \|\| err != nil {
	t.Fatalf("failed key conversion: pfx got %v want %v, id got %v want %v, gen got %v want %v, err %v", gotPfx, v.pfx, gotId, v.id, gotGen, v.gen, err)
	}
	}
	}