blob: 63ff62717a5bb02e9b6918e0ba833e94713e6958 [file] [log] [blame]
// 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
import (
wire ""
Test setup for normal conflicts:
Oid: x, isConflict = true, has local update, has remote update, has ancestor, resolution = pickLocal
Oid: b, isConflict = true, has local update, remote deleted, has ancestor, resolution = pickRemote
Oid: c, isConflict = true, local deleted, has remote update, has ancestor, resolution = pickRemote
Oid: p, isConflict = true, has local update, has remote update, has ancestor, resolution = createNew
Oid: y, isConflict = true, has local update, has remote update, has no ancestor, resolution = pickRemote
Oid: z, isConflict = false, no local update, has remote update, has unknown ancestor, resolution = pickRemote
Oid: e, isConflict = false, no local update, has remote update, has unknown ancestor, resolution = createNew
Oid: f, isConflict = false, no local update, has remote update, has unknown ancestor, resolution = pickLocal
Oid: g, isConflict = false, no local update, has remote update, has unknown ancestor, local head is deleted, resolution = pickLocal
Oid: a, local value rubberbanded in due to localBatch, resolution = createNew
localBatch1: {y, a}
remoteBatch1: {y, z, e, f}
Group 6:
Oid: la1, isConflict = true, has local update, has remote update, has ancestor, resolution = pickLocal
Oid: lb1, is a LinkLogRecord, isConflict = false, remote syncbase added a linklogrecord with local update as parent and remote update as the child, resolution = pickLocal
Oid: lc1, is a LinkLogRecord, isConflict = false, remote syncbase added a linklogrecord with local update as child and remote update as the parent, resolution = pickLocal
remoteBatch2: {la1, lb1, lc1}
Group 7:
Oid: la2, isConflict = true, has local update, has remote update, has ancestor, resolution = pickRemote
Oid: lb2, is a LinkLogRecord, isConflict = false, remote syncbase added a linklogrecord with local update as parent and remote update as the child, resolution = pickRemote
Oid: lc2, is a LinkLogRecord, isConflict = false, remote syncbase added a linklogrecord with local update as child and remote update as the parent, resolution = pickRemote
remoteBatch3: {la2, lb2, lc2}
Group 8:
Oid: la3, isConflict = true, has local update, has remote update, has ancestor, resolution = pickNew
Oid: lb3, is a LinkLogRecord, isConflict = false, remote syncbase added a linklogrecord with local update as parent and remote update as the child, resolution = pickNew
Oid: lc3, is a LinkLogRecord, isConflict = false, remote syncbase added a linklogrecord with local update as child and remote update as the parent, resolution = pickNew
remoteBatch4: {la3, lb3, lc3}
Group 9:
Oid: la4, isConflict = true, has local update, has remote update, has ancestor, resolution = pickLocal
Oid: lb4, isAddedByCr = true, isConflict = false, no remote update, no ancestor, resolution = pickLocal
Oid: lc4, is a LinkLogRecord, isConflict = false, remote syncbase added a linklogrecord with local update as child and remote update as the parent, resolution = pickLocal
localBatch5: {la4, lb4}
remoteBatch5: {la4, lc4}
var (
updObjectsAppResolves = map[string]*objConflictState{
// group1
x: createObjConflictState(true /*isConflict*/, true /*hasLocal*/, true /*hasRemote*/, true /*hasAncestor*/),
// group2
b: createObjConflictState(true /*isConflict*/, true /*hasLocal*/, true /*hasRemote*/, true /*hasAncestor*/),
// group3
c: createObjConflictState(true /*isConflict*/, true /*hasLocal*/, true /*hasRemote*/, true /*hasAncestor*/),
// group4
p: createObjConflictState(true /*isConflict*/, true /*hasLocal*/, true /*hasRemote*/, true /*hasAncestor*/),
// group5
y: createObjConflictState(true /*isConflict*/, true /*hasLocal*/, true /*hasRemote*/, false /*hasAncestor*/),
z: createObjConflictState(false /*isConflict*/, true /*hasLocal*/, true /*hasRemote*/, false /*hasAncestor*/),
a: createObjConflictState(false /*isConflict*/, true /*hasLocal*/, false /*hasRemote*/, false /*hasAncestor*/),
e: createObjConflictState(false /*isConflict*/, true /*hasLocal*/, true /*hasRemote*/, false /*hasAncestor*/),
f: createObjConflictState(false /*isConflict*/, true /*hasLocal*/, true /*hasRemote*/, false /*hasAncestor*/),
g: createObjConflictState(false /*isConflict*/, true /*hasLocal*/, true /*hasRemote*/, false /*hasAncestor*/),
// group6
la1: createObjConflictState(true /*isConflict*/, true /*hasLocal*/, true /*hasRemote*/, true /*hasAncestor*/),
lb1: createObjLinkState(true /*isRemoteBlessed*/),
lc1: createObjLinkState(false /*isRemoteBlessed*/),
// group7
la2: createObjConflictState(true /*isConflict*/, true /*hasLocal*/, true /*hasRemote*/, true /*hasAncestor*/),
lb2: createObjLinkState(true /*isRemoteBlessed*/),
lc2: createObjLinkState(false /*isRemoteBlessed*/),
// group8
la3: createObjConflictState(true /*isConflict*/, true /*hasLocal*/, true /*hasRemote*/, true /*hasAncestor*/),
lb3: createObjLinkState(true /*isRemoteBlessed*/),
lc3: createObjLinkState(false /*isRemoteBlessed*/),
// group9
la4: createObjConflictState(true /*isConflict*/, true /*hasLocal*/, true /*hasRemote*/, true /*hasAncestor*/),
lb4: createObjConflictState(false /*isConflict*/, true /*hasLocal*/, false /*hasRemote*/, false /*hasAncestor*/),
lc4: createObjLinkState(false /*isRemoteBlessed*/),
localBatchId1 uint64 = 34
remoteBatchId1 uint64 = 58
remoteBatchId2 uint64 = 72
remoteBatchId3 uint64 = 45
remoteBatchId4 uint64 = 23
localBatchId5 uint64 = 98
remoteBatchId5 uint64 = 78
func createGroupedCrTestData() *groupedCrData {
groupedCrTestData := &groupedCrData{oids: map[string]bool{}}
var group *crGroup
// group1
group = newGroup()
addToGroup(group, x, NoBatchId, -1)
groupedCrTestData.oids[x] = true
groupedCrTestData.groups = append(groupedCrTestData.groups, group)
// group2
group = newGroup()
addToGroup(group, b, NoBatchId, -1)
groupedCrTestData.oids[b] = true
groupedCrTestData.groups = append(groupedCrTestData.groups, group)
// group3
group = newGroup()
addToGroup(group, c, NoBatchId, -1)
groupedCrTestData.oids[c] = true
groupedCrTestData.groups = append(groupedCrTestData.groups, group)
// group4
group = newGroup()
addToGroup(group, p, NoBatchId, -1)
groupedCrTestData.oids[p] = true
groupedCrTestData.groups = append(groupedCrTestData.groups, group)
// group5
group = newGroup()
addToGroup(group, y, localBatchId1, wire.BatchSourceLocal)
addToGroup(group, y, remoteBatchId1, wire.BatchSourceRemote)
addToGroup(group, z, remoteBatchId1, wire.BatchSourceRemote)
addToGroup(group, e, remoteBatchId1, wire.BatchSourceRemote)
addToGroup(group, f, remoteBatchId1, wire.BatchSourceRemote)
addToGroup(group, g, remoteBatchId1, wire.BatchSourceRemote)
addToGroup(group, a, localBatchId1, wire.BatchSourceLocal)
groupedCrTestData.oids[y] = true
groupedCrTestData.oids[z] = true
groupedCrTestData.oids[a] = true
groupedCrTestData.oids[e] = true
groupedCrTestData.oids[f] = true
groupedCrTestData.groups = append(groupedCrTestData.groups, group)
// group6
group = newGroup()
addToGroup(group, la1, remoteBatchId2, wire.BatchSourceRemote)
addToGroup(group, lb1, remoteBatchId2, wire.BatchSourceRemote)
addToGroup(group, lc1, remoteBatchId2, wire.BatchSourceRemote)
groupedCrTestData.oids[la1] = true
groupedCrTestData.oids[lb1] = true
groupedCrTestData.oids[lc1] = true
groupedCrTestData.groups = append(groupedCrTestData.groups, group)
// group7
group = newGroup()
addToGroup(group, la2, remoteBatchId3, wire.BatchSourceRemote)
addToGroup(group, lb2, remoteBatchId3, wire.BatchSourceRemote)
addToGroup(group, lc2, remoteBatchId3, wire.BatchSourceRemote)
groupedCrTestData.oids[la2] = true
groupedCrTestData.oids[lb2] = true
groupedCrTestData.oids[lc2] = true
groupedCrTestData.groups = append(groupedCrTestData.groups, group)
// group8
group = newGroup()
addToGroup(group, la3, remoteBatchId4, wire.BatchSourceRemote)
addToGroup(group, lb3, remoteBatchId4, wire.BatchSourceRemote)
addToGroup(group, lc3, remoteBatchId4, wire.BatchSourceRemote)
groupedCrTestData.oids[la3] = true
groupedCrTestData.oids[lb3] = true
groupedCrTestData.oids[lc3] = true
groupedCrTestData.groups = append(groupedCrTestData.groups, group)
// group9
group = newGroup()
addToGroup(group, la4, remoteBatchId5, wire.BatchSourceRemote)
addToGroup(group, la4, localBatchId5, wire.BatchSourceLocal)
addToGroup(group, lb4, localBatchId5, wire.BatchSourceLocal)
addToGroup(group, lc4, remoteBatchId5, wire.BatchSourceRemote)
groupedCrTestData.oids[la4] = true
groupedCrTestData.oids[lb4] = true
groupedCrTestData.oids[lc4] = true
groupedCrTestData.groups = append(groupedCrTestData.groups, group)
return groupedCrTestData
func createAndSaveNodeAndLogRecDataAppResolves(iSt *initiationState) {
// group1
saveNodeAndLogRec(iSt.tx, x, updObjectsAppResolves[x].oldHead, 24, false)
saveNodeAndLogRec(iSt.tx, x, updObjectsAppResolves[x].newHead, 25, false)
saveNodeAndLogRec(iSt.tx, x, updObjectsAppResolves[x].ancestor, 20, false)
// group2
saveNodeAndLogRec(iSt.tx, b, updObjectsAppResolves[b].oldHead, 56, false)
saveNodeAndLogRec(iSt.tx, b, updObjectsAppResolves[b].newHead, 23, true)
saveNodeAndLogRec(iSt.tx, b, updObjectsAppResolves[b].ancestor, 15, false)
// group3
saveNodeAndLogRec(iSt.tx, c, updObjectsAppResolves[c].oldHead, 56, true)
saveNodeAndLogRec(iSt.tx, c, updObjectsAppResolves[c].newHead, 23, false)
saveNodeAndLogRec(iSt.tx, c, updObjectsAppResolves[c].ancestor, 15, false)
// group4
saveNodeAndLogRec(iSt.tx, p, updObjectsAppResolves[p].oldHead, 56, false)
saveNodeAndLogRec(iSt.tx, p, updObjectsAppResolves[p].newHead, 23, false)
saveNodeAndLogRec(iSt.tx, p, updObjectsAppResolves[p].ancestor, 15, false)
saveNodeAndLogRec(iSt.tx, y, updObjectsAppResolves[y].oldHead, 56, false)
saveNodeAndLogRec(iSt.tx, y, updObjectsAppResolves[y].newHead, 23, false)
saveNodeAndLogRec(iSt.tx, z, updObjectsAppResolves[z].oldHead, 56, false)
saveNodeAndLogRec(iSt.tx, z, updObjectsAppResolves[z].newHead, 23, false)
saveNodeAndLogRec(iSt.tx, e, updObjectsAppResolves[e].oldHead, 56, false)
saveNodeAndLogRec(iSt.tx, e, updObjectsAppResolves[e].newHead, 23, false)
saveNodeAndLogRec(iSt.tx, f, updObjectsAppResolves[f].oldHead, 56, false)
saveNodeAndLogRec(iSt.tx, f, updObjectsAppResolves[f].newHead, 23, false)
saveNodeAndLogRec(iSt.tx, g, updObjectsAppResolves[g].oldHead, 56, true)
saveNodeAndLogRec(iSt.tx, g, updObjectsAppResolves[g].newHead, 23, false)
saveNodeAndLogRec(iSt.tx, a, updObjectsAppResolves[a].oldHead, 56, false)
// group6
saveNodeAndLogRec(iSt.tx, la1, updObjectsAppResolves[la1].oldHead, 56, false)
saveNodeAndLogRec(iSt.tx, la1, updObjectsAppResolves[la1].newHead, 23, false)
saveNodeAndLogRec(iSt.tx, la1, updObjectsAppResolves[la1].ancestor, 15, false)
saveNodeAndLogRec(iSt.tx, lb1, updObjectsAppResolves[lb1].oldHead, 56, false)
saveNodeAndLogRec(iSt.tx, lb1, updObjectsAppResolves[lb1].newHead, 23, false)
saveNodeAndLogRec(iSt.tx, lc1, updObjectsAppResolves[lc1].oldHead, 56, false)
// lc1.newHead is same as lc1.oldHead.
// group7
saveNodeAndLogRec(iSt.tx, la2, updObjectsAppResolves[la2].oldHead, 56, false)
saveNodeAndLogRec(iSt.tx, la2, updObjectsAppResolves[la2].newHead, 23, false)
saveNodeAndLogRec(iSt.tx, la2, updObjectsAppResolves[la2].ancestor, 15, false)
saveNodeAndLogRec(iSt.tx, lb2, updObjectsAppResolves[lb2].oldHead, 56, false)
saveNodeAndLogRec(iSt.tx, lb2, updObjectsAppResolves[lb2].newHead, 23, false)
saveNodeAndLogRec(iSt.tx, lc2, updObjectsAppResolves[lc2].oldHead, 56, false)
// lc2.newHead is same as lc2.oldHead.
// group8
saveNodeAndLogRec(iSt.tx, la3, updObjectsAppResolves[la3].oldHead, 56, false)
saveNodeAndLogRec(iSt.tx, la3, updObjectsAppResolves[la3].newHead, 23, false)
saveNodeAndLogRec(iSt.tx, la3, updObjectsAppResolves[la3].ancestor, 15, false)
saveNodeAndLogRec(iSt.tx, lb3, updObjectsAppResolves[lb3].oldHead, 56, false)
saveNodeAndLogRec(iSt.tx, lb3, updObjectsAppResolves[lb3].newHead, 23, false)
saveNodeAndLogRec(iSt.tx, lc3, updObjectsAppResolves[lc3].oldHead, 56, false)
// lc3.newHead is same as lc3.oldHead.
// group8
saveNodeAndLogRec(iSt.tx, la4, updObjectsAppResolves[la4].oldHead, 56, false)
saveNodeAndLogRec(iSt.tx, la4, updObjectsAppResolves[la4].newHead, 23, false)
saveNodeAndLogRec(iSt.tx, la4, updObjectsAppResolves[la4].ancestor, 15, false)
saveNodeAndLogRec(iSt.tx, lb4, updObjectsAppResolves[lb4].oldHead, 56, false)
// lb4.newHead is NoVersion
saveNodeAndLogRec(iSt.tx, lc4, updObjectsAppResolves[lc4].oldHead, 56, false)
// lc4.newHead is same as lc4.oldHead.
func writeVersionedValues(t *testing.T, iSt *initiationState) {
// group1
saveValue(t, iSt.tx, x, updObjectsAppResolves[x].oldHead)
saveValue(t, iSt.tx, x, updObjectsAppResolves[x].newHead)
saveValue(t, iSt.tx, x, updObjectsAppResolves[x].ancestor)
// group2
saveValue(t, iSt.tx, b, updObjectsAppResolves[b].oldHead)
saveValue(t, iSt.tx, b, updObjectsAppResolves[b].ancestor)
// group3
saveValue(t, iSt.tx, c, updObjectsAppResolves[c].newHead)
saveValue(t, iSt.tx, c, updObjectsAppResolves[c].ancestor)
// group4
saveValue(t, iSt.tx, p, updObjectsAppResolves[p].oldHead)
saveValue(t, iSt.tx, p, updObjectsAppResolves[p].newHead)
saveValue(t, iSt.tx, p, updObjectsAppResolves[p].ancestor)
// group5
saveValue(t, iSt.tx, y, updObjectsAppResolves[y].oldHead)
saveValue(t, iSt.tx, y, updObjectsAppResolves[y].newHead)
saveValue(t, iSt.tx, z, updObjectsAppResolves[z].oldHead)
saveValue(t, iSt.tx, z, updObjectsAppResolves[z].newHead)
saveValue(t, iSt.tx, e, updObjectsAppResolves[e].oldHead)
saveValue(t, iSt.tx, e, updObjectsAppResolves[e].newHead)
saveValue(t, iSt.tx, f, updObjectsAppResolves[f].oldHead)
saveValue(t, iSt.tx, f, updObjectsAppResolves[f].newHead)
// No value for oldHead for g as g was deleted on local.
saveValue(t, iSt.tx, g, updObjectsAppResolves[g].newHead)
saveValue(t, iSt.tx, a, updObjectsAppResolves[a].oldHead)
// group6
saveValue(t, iSt.tx, la1, updObjectsAppResolves[la1].oldHead)
saveValue(t, iSt.tx, la1, updObjectsAppResolves[la1].newHead)
saveValue(t, iSt.tx, la1, updObjectsAppResolves[la1].ancestor)
saveValue(t, iSt.tx, lb1, updObjectsAppResolves[lb1].oldHead)
saveValue(t, iSt.tx, lb1, updObjectsAppResolves[lb1].newHead)
saveValue(t, iSt.tx, lc1, updObjectsAppResolves[lc1].oldHead)
// group7
saveValue(t, iSt.tx, la2, updObjectsAppResolves[la2].oldHead)
saveValue(t, iSt.tx, la2, updObjectsAppResolves[la2].newHead)
saveValue(t, iSt.tx, la2, updObjectsAppResolves[la2].ancestor)
saveValue(t, iSt.tx, lb2, updObjectsAppResolves[lb2].oldHead)
saveValue(t, iSt.tx, lb2, updObjectsAppResolves[lb2].newHead)
saveValue(t, iSt.tx, lc2, updObjectsAppResolves[lc2].oldHead)
// group8
saveValue(t, iSt.tx, la3, updObjectsAppResolves[la3].oldHead)
saveValue(t, iSt.tx, la3, updObjectsAppResolves[la3].newHead)
saveValue(t, iSt.tx, la3, updObjectsAppResolves[la3].ancestor)
saveValue(t, iSt.tx, lb3, updObjectsAppResolves[lb3].oldHead)
saveValue(t, iSt.tx, lb3, updObjectsAppResolves[lb3].newHead)
saveValue(t, iSt.tx, lc3, updObjectsAppResolves[lc3].oldHead)
// group9
saveValue(t, iSt.tx, la4, updObjectsAppResolves[la4].oldHead)
saveValue(t, iSt.tx, la4, updObjectsAppResolves[la4].newHead)
saveValue(t, iSt.tx, la4, updObjectsAppResolves[la4].ancestor)
saveValue(t, iSt.tx, lb4, updObjectsAppResolves[lb4].oldHead)
saveValue(t, iSt.tx, lc4, updObjectsAppResolves[lc4].oldHead)
func setResInfoData(mockCrs *conflictResolverStream) {
var newValue *vom.RawBytes
newValue, _ = vom.RawBytesFromValue("newValue")
// group1
addResInfo(mockCrs, x, wire.ValueSelectionLocal, nil, false)
// group2
addResInfo(mockCrs, b, wire.ValueSelectionRemote, nil, false)
// group3
addResInfo(mockCrs, c, wire.ValueSelectionRemote, nil, false)
// group4
addResInfo(mockCrs, p, wire.ValueSelectionOther, newValue, false)
// group5
addResInfo(mockCrs, y, wire.ValueSelectionRemote, nil, true)
addResInfo(mockCrs, z, wire.ValueSelectionRemote, nil, true)
addResInfo(mockCrs, e, wire.ValueSelectionOther, newValue, true)
addResInfo(mockCrs, f, wire.ValueSelectionLocal, nil, true)
addResInfo(mockCrs, g, wire.ValueSelectionLocal, nil, true)
addResInfo(mockCrs, a, wire.ValueSelectionOther, newValue, false)
// group6
addResInfo(mockCrs, la1, wire.ValueSelectionLocal, nil, true)
addResInfo(mockCrs, lb1, wire.ValueSelectionLocal, nil, true)
addResInfo(mockCrs, lc1, wire.ValueSelectionLocal, nil, false)
// group7
addResInfo(mockCrs, la2, wire.ValueSelectionRemote, nil, true)
addResInfo(mockCrs, lb2, wire.ValueSelectionRemote, nil, true)
addResInfo(mockCrs, lc2, wire.ValueSelectionRemote, nil, false)
// group8
addResInfo(mockCrs, la3, wire.ValueSelectionOther, newValue, true)
addResInfo(mockCrs, lb3, wire.ValueSelectionOther, newValue, true)
addResInfo(mockCrs, lc3, wire.ValueSelectionOther, newValue, false)
// group9
addResInfo(mockCrs, la4, wire.ValueSelectionLocal, nil, true)
addResInfo(mockCrs, lb4, wire.ValueSelectionLocal, nil, true)
addResInfo(mockCrs, lc4, wire.ValueSelectionLocal, nil, false)
func TestResolveViaApp(t *testing.T) {
service := createService(t)
defer destroyService(t, service)
db := createDatabase(t, service)
iSt := &initiationState{
updObjects: updObjectsAppResolves,
tx: db.St().NewWatchableTransaction(),
config: &initiationConfig{
sync: service.sync,
db: db,
writeVersionedValues(t, iSt)
mockCrs := &conflictResolverStream{}
groupedCrTestData := createGroupedCrTestData()
if err := iSt.resolveViaApp(nil, groupedCrTestData); err != nil {
t.Errorf("Error returned by resolveViaApp: %v", err)
// verify
verifyConflictInfos(t, mockCrs)
// group1
verifyResolution(t, iSt.updObjects, x, pickLocal)
// group2
verifyResolution(t, iSt.updObjects, b, pickRemote)
// group3
verifyResolution(t, iSt.updObjects, c, pickRemote)
// group4
verifyCreateNew(t, iSt.updObjects, p, false)
// group5
verifyResolution(t, iSt.updObjects, y, pickRemote)
verifyResolution(t, iSt.updObjects, z, pickRemote)
verifyCreateNew(t, iSt.updObjects, e, false)
verifyCreateNew(t, iSt.updObjects, f, false)
verifyCreateNew(t, iSt.updObjects, g, true)
verifyCreateNew(t, iSt.updObjects, a, false)
// group6
verifyResolution(t, iSt.updObjects, la1, pickLocal)
// to avoid circular loop in dag pickLocal is converted to createNew
verifyCreateNew(t, iSt.updObjects, lb1, false)
verifyResolution(t, iSt.updObjects, lc1, pickLocal)
// group7
verifyResolution(t, iSt.updObjects, la2, pickRemote)
verifyResolution(t, iSt.updObjects, lb2, pickRemote)
// since localHead == remoteHead, the response is optimized and converted to
// pickLocal
verifyResolution(t, iSt.updObjects, lc2, pickLocal)
// group8
verifyCreateNew(t, iSt.updObjects, la3, false)
verifyCreateNew(t, iSt.updObjects, lb3, false)
verifyCreateNew(t, iSt.updObjects, lc3, false)
// group9
verifyResolution(t, iSt.updObjects, la4, pickLocal)
verifyResolution(t, iSt.updObjects, lb4, pickLocal)
verifyResolution(t, iSt.updObjects, lc4, pickLocal)
// verify batch ids
verifyBatchId(t, iSt.updObjects, NoBatchId, x, b, c, p)
bid := iSt.updObjects[y].res.batchId
if bid == NoBatchId {
t.Errorf("BatchId for group5 should not be NoBatchId")
verifyBatchId(t, iSt.updObjects, bid, y, z, e, f, a)
bid = iSt.updObjects[la1].res.batchId
if bid == NoBatchId {
t.Errorf("BatchId for group6 should not be NoBatchId")
verifyBatchId(t, iSt.updObjects, bid, la1, lb1, lc1)
bid = iSt.updObjects[la2].res.batchId
if bid == NoBatchId {
t.Errorf("BatchId for group7 should not be NoBatchId")
verifyBatchId(t, iSt.updObjects, bid, la2, lb2, lc2)
bid = iSt.updObjects[la3].res.batchId
if bid == NoBatchId {
t.Errorf("BatchId for group8 should not be NoBatchId")
verifyBatchId(t, iSt.updObjects, bid, la3, lb3, lc3)
bid = iSt.updObjects[la4].res.batchId
if bid == NoBatchId {
t.Errorf("BatchId for group9 should not be NoBatchId")
verifyBatchId(t, iSt.updObjects, bid, la4, lb4, lc4)
func verifyConflictInfos(t *testing.T, mockCrs *conflictResolverStream) {
var ci wire.ConflictInfo
if len(mockCrs.sendQ) != 29 {
t.Errorf("ConflictInfo count expected: %v, actual: %v", 29, len(mockCrs.sendQ))
// group1
ci = mockCrs.sendQ[0]
checkConflictRow(t, x, ci, []uint64{}, false /*localDeleted*/, false /*remoteDeleted*/)
checkContinued(t, mockCrs.sendQ[0:1])
// group2
ci = mockCrs.sendQ[1]
checkConflictRow(t, b, ci, []uint64{}, false /*localDeleted*/, true /*remoteDeleted*/)
checkContinued(t, mockCrs.sendQ[1:2])
// group3
ci = mockCrs.sendQ[2]
checkConflictRow(t, c, ci, []uint64{}, true /*localDeleted*/, false /*remoteDeleted*/)
checkContinued(t, mockCrs.sendQ[2:3])
// group4
ci = mockCrs.sendQ[3]
checkConflictRow(t, p, ci, []uint64{}, false /*localDeleted*/, false /*remoteDeleted*/)
checkContinued(t, mockCrs.sendQ[3:4])
// group5
batchMap := map[uint64]wire.ConflictInfo{}
batchMap[getBid(mockCrs.sendQ[4])] = mockCrs.sendQ[4]
batchMap[getBid(mockCrs.sendQ[5])] = mockCrs.sendQ[5]
ci = batchMap[localBatchId1]
checkConflictBatch(t, localBatchId1, ci, wire.BatchSourceLocal)
ci = batchMap[remoteBatchId1]
checkConflictBatch(t, remoteBatchId1, ci, wire.BatchSourceRemote)
rowMap := map[string]wire.ConflictInfo{}
rowMap[getOid(mockCrs.sendQ[6])] = mockCrs.sendQ[6]
rowMap[getOid(mockCrs.sendQ[7])] = mockCrs.sendQ[7]
rowMap[getOid(mockCrs.sendQ[8])] = mockCrs.sendQ[8]
rowMap[getOid(mockCrs.sendQ[9])] = mockCrs.sendQ[9]
rowMap[getOid(mockCrs.sendQ[10])] = mockCrs.sendQ[10]
rowMap[getOid(mockCrs.sendQ[11])] = mockCrs.sendQ[11]
ci = rowMap[y]
checkConflictRow(t, y, ci, []uint64{localBatchId1, remoteBatchId1}, false /*localDeleted*/, false /*remoteDeleted*/)
ci = rowMap[z]
checkConflictRow(t, z, ci, []uint64{remoteBatchId1}, false /*localDeleted*/, false /*remoteDeleted*/)
ci = rowMap[a]
checkConflictRow(t, a, ci, []uint64{localBatchId1}, false /*localDeleted*/, false /*remoteDeleted*/)
ci = rowMap[e]
checkConflictRow(t, e, ci, []uint64{remoteBatchId1}, false /*localDeleted*/, false /*remoteDeleted*/)
ci = rowMap[f]
checkConflictRow(t, f, ci, []uint64{remoteBatchId1}, false /*localDeleted*/, false /*remoteDeleted*/)
ci = rowMap[g]
checkConflictRow(t, g, ci, []uint64{remoteBatchId1}, true /*localDeleted*/, false /*remoteDeleted*/)
checkContinued(t, mockCrs.sendQ[4:12])
// group6
ci = mockCrs.sendQ[12]
checkConflictBatch(t, remoteBatchId2, ci, wire.BatchSourceRemote)
rowMap[getOid(mockCrs.sendQ[13])] = mockCrs.sendQ[13]
rowMap[getOid(mockCrs.sendQ[14])] = mockCrs.sendQ[14]
rowMap[getOid(mockCrs.sendQ[15])] = mockCrs.sendQ[15]
ci = rowMap[la1]
checkConflictRow(t, la1, ci, []uint64{remoteBatchId2}, false /*localDeleted*/, false /*remoteDeleted*/)
ci = rowMap[lb1]
checkConflictRow(t, lb1, ci, []uint64{remoteBatchId2}, false /*localDeleted*/, false /*remoteDeleted*/)
ci = rowMap[lc1]
checkConflictRow(t, lc1, ci, []uint64{remoteBatchId2}, false /*localDeleted*/, false /*remoteDeleted*/)
checkContinued(t, mockCrs.sendQ[12:16])
// group7
ci = mockCrs.sendQ[16]
checkConflictBatch(t, remoteBatchId3, ci, wire.BatchSourceRemote)
rowMap[getOid(mockCrs.sendQ[17])] = mockCrs.sendQ[17]
rowMap[getOid(mockCrs.sendQ[18])] = mockCrs.sendQ[18]
rowMap[getOid(mockCrs.sendQ[19])] = mockCrs.sendQ[19]
ci = rowMap[la2]
checkConflictRow(t, la2, ci, []uint64{remoteBatchId3}, false /*localDeleted*/, false /*remoteDeleted*/)
ci = rowMap[lb2]
checkConflictRow(t, lb2, ci, []uint64{remoteBatchId3}, false /*localDeleted*/, false /*remoteDeleted*/)
ci = rowMap[lc2]
checkConflictRow(t, lc2, ci, []uint64{remoteBatchId3}, false /*localDeleted*/, false /*remoteDeleted*/)
checkContinued(t, mockCrs.sendQ[16:20])
// group8
ci = mockCrs.sendQ[20]
checkConflictBatch(t, remoteBatchId4, ci, wire.BatchSourceRemote)
rowMap[getOid(mockCrs.sendQ[21])] = mockCrs.sendQ[21]
rowMap[getOid(mockCrs.sendQ[22])] = mockCrs.sendQ[22]
rowMap[getOid(mockCrs.sendQ[23])] = mockCrs.sendQ[23]
ci = rowMap[la3]
checkConflictRow(t, la3, ci, []uint64{remoteBatchId4}, false /*localDeleted*/, false /*remoteDeleted*/)
ci = rowMap[lb3]
checkConflictRow(t, lb3, ci, []uint64{remoteBatchId4}, false /*localDeleted*/, false /*remoteDeleted*/)
ci = rowMap[lc3]
checkConflictRow(t, lc3, ci, []uint64{remoteBatchId4}, false /*localDeleted*/, false /*remoteDeleted*/)
checkContinued(t, mockCrs.sendQ[20:24])
// group9
batchMap = map[uint64]wire.ConflictInfo{}
batchMap[getBid(mockCrs.sendQ[24])] = mockCrs.sendQ[24]
batchMap[getBid(mockCrs.sendQ[25])] = mockCrs.sendQ[25]
ci = batchMap[localBatchId5]
checkConflictBatch(t, localBatchId5, ci, wire.BatchSourceLocal)
ci = batchMap[remoteBatchId5]
checkConflictBatch(t, remoteBatchId5, ci, wire.BatchSourceRemote)
rowMap[getOid(mockCrs.sendQ[26])] = mockCrs.sendQ[26]
rowMap[getOid(mockCrs.sendQ[27])] = mockCrs.sendQ[27]
rowMap[getOid(mockCrs.sendQ[28])] = mockCrs.sendQ[28]
ci = rowMap[la4]
checkConflictRow(t, la4, ci, []uint64{localBatchId5, remoteBatchId5}, false /*localDeleted*/, false /*remoteDeleted*/)
ci = rowMap[lb4]
checkConflictRow(t, lb4, ci, []uint64{localBatchId5}, false /*localDeleted*/, false /*remoteDeleted*/)
ci = rowMap[lc4]
checkConflictRow(t, lc4, ci, []uint64{remoteBatchId5}, false /*localDeleted*/, false /*remoteDeleted*/)
checkContinued(t, mockCrs.sendQ[24:])
func checkContinued(t *testing.T, infoGroup []wire.ConflictInfo) {
lastIndex := len(infoGroup) - 1
for i := range infoGroup {
if infoGroup[i].Continued != (i != lastIndex) {
t.Errorf("Wrong value for continued field in %#v", infoGroup[i])
func getOid(ci wire.ConflictInfo) string {
ciData := ci.Data.(wire.ConflictDataRow).Value
writeOp := ciData.Op.(wire.OperationWrite).Value
return toRowKey(writeOp.Key)
func getBid(ci wire.ConflictInfo) uint64 {
ciData := ci.Data.(wire.ConflictDataBatch).Value
return ciData.Id
func checkConflictBatch(t *testing.T, bid uint64, ci wire.ConflictInfo, source wire.BatchSource) {
ciData := ci.Data.(wire.ConflictDataBatch).Value
if ciData.Source != source {
t.Errorf("Source for batchid %v expected: %#v, actual: %#v", bid, source, ciData.Source)
if ci.Continued != true {
t.Errorf("Bid: %v, Unexpected value for continued: %v", bid, ci.Continued)
func checkConflictRow(t *testing.T, oid string, ci wire.ConflictInfo, batchIds []uint64, localDeleted, remoteDeleted bool) {
ciData := ci.Data.(wire.ConflictDataRow).Value
writeOp := ciData.Op.(wire.OperationWrite).Value
st, _ := updObjectsAppResolves[oid]
var err error
var ancestorValueBytes []byte
if writeOp.AncestorValue.Bytes != nil {
if err = writeOp.AncestorValue.Bytes.ToValue(&ancestorValueBytes); err != nil {
t.Errorf("Oid: %v: error decoding writeOp.AncestorValue.Bytes: %v", oid, err)
if (st.ancestor != NoVersion) && !bytes.Equal(makeValue(oid, st.ancestor), ancestorValueBytes) {
t.Errorf("Oid: %v, Ancestor value expected: %v, actual: %v", oid, string(makeValue(oid, st.ancestor)), string(ancestorValueBytes))
if remoteDeleted && writeOp.RemoteValue == nil {
t.Errorf("Oid: %v, for remote deleted remote value is expected to have an instance with no bytes", oid)
var remoteValueBytes []byte
if writeOp.RemoteValue.Bytes != nil {
if err = writeOp.RemoteValue.Bytes.ToValue(&remoteValueBytes); err != nil {
t.Errorf("Oid: %v: error decoding writeOp.RemoteValue.Bytes: %v", oid, err)
if !remoteDeleted && (st.newHead != NoVersion) && !bytes.Equal(makeValue(oid, st.newHead), remoteValueBytes) {
t.Errorf("Oid: %v, Remote value expected: %v, actual: %v", oid, string(makeValue(oid, st.newHead)), string(remoteValueBytes))
if localDeleted && writeOp.LocalValue == nil {
t.Errorf("Oid: %v, for local deleted local value is expected to have an instance with no bytes", oid)
var localValueBytes []byte
if writeOp.LocalValue.Bytes != nil {
if err = writeOp.LocalValue.Bytes.ToValue(&localValueBytes); err != nil {
t.Errorf("Oid: %v: error decoding writeOp.LocalValue.Bytes: %v", oid, err)
if !localDeleted && (st.oldHead != NoVersion) && !bytes.Equal(makeValue(oid, st.oldHead), localValueBytes) {
t.Errorf("Oid: %v, Local value expected: %v, actual: %v", oid, string(makeValue(oid, st.oldHead)), string(localValueBytes))
if !uint64ArrayEq(batchIds, ciData.BatchIds) {
t.Errorf("Oid: %v, BatchIds expected: %v, actual: %v", oid, batchIds, ciData.BatchIds)
func addResInfo(crs *conflictResolverStream, oid string, sel wire.ValueSelection, val *vom.RawBytes, cntd bool) {
var valRes *wire.Value
if val != nil {
valRes = &wire.Value{
Bytes: val,
rInfo := wire.ResolutionInfo{
Key: common.StripFirstKeyPartOrDie(oid),
Selection: sel,
Result: valRes,
Continued: cntd,
crs.recvQ = append(crs.recvQ, rInfo)
func saveValue(t *testing.T, tx *watchable.Transaction, oid, version string) {
encodedValue, err := vom.Encode(makeValue(oid, version))
if err != nil {
t.Errorf("Error encoding %s,%s: %v", oid, version, err)
if err := watchable.PutAtVersion(nil, tx, []byte(oid), encodedValue, []byte(version)); err != nil {
t.Errorf("Failed to write versioned value for oid,ver: %s,%s", oid, version)
func makeValue(oid, ver string) []byte {
return []byte(oid + ver)
func addToGroup(group *crGroup, oid string, bid uint64, source wire.BatchSource) {
if bid != NoBatchId {
group.batchSource[bid] = source
group.batchesByOid[oid] = append(group.batchesByOid[oid], bid)
} else {
group.batchesByOid[oid] = []uint64{}
func newGroup() *crGroup {
return &crGroup{
batchSource: map[uint64]wire.BatchSource{},
batchesByOid: map[string][]uint64{},