Add Conflict resolution API
1) Client side API for receiving Conflicts
2) Model objects for Conflict and Resolution data
3) RPC API to register conflict resolver with syncbase service
4) Example conflict resolver along with tests to verify it
Change-Id: I671bcadd30b12c0388664ff7c8d315079eae7ddb
diff --git a/v23/services/syncbase/nosql/service.vdl b/v23/services/syncbase/nosql/service.vdl
index ebeccf5..16182de 100644
--- a/v23/services/syncbase/nosql/service.vdl
+++ b/v23/services/syncbase/nosql/service.vdl
@@ -73,6 +73,10 @@
// SchemaManager implements the API for managing schema metadata attached
// to a Database.
SchemaManager
+
+ // ConflictManager implements the API for registering resolvers, receiving
+ // conflicts and sending resolutions.
+ ConflictManager
}
// Table represents a collection of Rows.
@@ -234,6 +238,30 @@
SetSchemaMetadata(metadata SchemaMetadata) error {access.Write}
}
+// ConflictManager interface provides all the methods necessary to handle
+// conflict resolution for a given database.
+type ConflictManager interface {
+ // StartConflictResolver registers a resolver for the database that is
+ // associated with this ConflictManager and creates a stream to receive
+ // conflicts and send resolutions.
+ // Batches of ConflictInfos will be sent over with the Continued field
+ // within the ConflictInfo representing the batch boundary. Client must
+ // respond with a batch of ResolutionInfos in the same fashion.
+ // A key is under conflict if two different values were written to it
+ // concurrently (in logical time), i.e. neither value is an ancestor of the
+ // other in the history graph.
+ // A key under conflict can be a part of a batch committed on local or
+ // remote or both syncbases. ConflictInfos for all keys in these two batches
+ // are grouped together. These keys may themselves be under conflict; the
+ // presented batch is a transitive closure of all batches containing keys
+ // under conflict.
+ // For example, for local batch {key1, key2} and remote batch {key1, key3},
+ // the batch sent for conflict resolution will be {key1, key2, key3}.
+ // If there was another concurrent batch {key2, key4}, then the batch sent
+ // for conflict resolution will be {key1, key2, key3, key4}.
+ StartConflictResolver() stream<ResolutionInfo, ConflictInfo> error {access.Write}
+}
+
// BlobManager is the interface for blob operations.
type BlobManager interface {
// API for resumable blob creation (append-only). After commit, a blob
diff --git a/v23/services/syncbase/nosql/service.vdl.go b/v23/services/syncbase/nosql/service.vdl.go
index 2d3d4ba..a908cfa 100644
--- a/v23/services/syncbase/nosql/service.vdl.go
+++ b/v23/services/syncbase/nosql/service.vdl.go
@@ -1452,6 +1452,356 @@
},
}
+// ConflictManagerClientMethods is the client interface
+// containing ConflictManager methods.
+//
+// ConflictManager interface provides all the methods necessary to handle
+// conflict resolution for a given database.
+type ConflictManagerClientMethods interface {
+ // StartConflictResolver registers a resolver for the database that is
+ // associated with this ConflictManager and creates a stream to receive
+ // conflicts and send resolutions.
+ // Batches of ConflictInfos will be sent over with the Continued field
+ // within the ConflictInfo representing the batch boundary. Client must
+ // respond with a batch of ResolutionInfos in the same fashion.
+ // A key is under conflict if two different values were written to it
+ // concurrently (in logical time), i.e. neither value is an ancestor of the
+ // other in the history graph.
+ // A key under conflict can be a part of a batch committed on local or
+ // remote or both syncbases. ConflictInfos for all keys in these two batches
+ // are grouped together. These keys may themselves be under conflict; the
+ // presented batch is a transitive closure of all batches containing keys
+ // under conflict.
+ // For example, for local batch {key1, key2} and remote batch {key1, key3},
+ // the batch sent for conflict resolution will be {key1, key2, key3}.
+ // If there was another concurrent batch {key2, key4}, then the batch sent
+ // for conflict resolution will be {key1, key2, key3, key4}.
+ StartConflictResolver(*context.T, ...rpc.CallOpt) (ConflictManagerStartConflictResolverClientCall, error)
+}
+
+// ConflictManagerClientStub adds universal methods to ConflictManagerClientMethods.
+type ConflictManagerClientStub interface {
+ ConflictManagerClientMethods
+ rpc.UniversalServiceMethods
+}
+
+// ConflictManagerClient returns a client stub for ConflictManager.
+func ConflictManagerClient(name string) ConflictManagerClientStub {
+ return implConflictManagerClientStub{name}
+}
+
+type implConflictManagerClientStub struct {
+ name string
+}
+
+func (c implConflictManagerClientStub) StartConflictResolver(ctx *context.T, opts ...rpc.CallOpt) (ocall ConflictManagerStartConflictResolverClientCall, err error) {
+ var call rpc.ClientCall
+ if call, err = v23.GetClient(ctx).StartCall(ctx, c.name, "StartConflictResolver", nil, opts...); err != nil {
+ return
+ }
+ ocall = &implConflictManagerStartConflictResolverClientCall{ClientCall: call}
+ return
+}
+
+// ConflictManagerStartConflictResolverClientStream is the client stream for ConflictManager.StartConflictResolver.
+type ConflictManagerStartConflictResolverClientStream interface {
+ // RecvStream returns the receiver side of the ConflictManager.StartConflictResolver client stream.
+ RecvStream() interface {
+ // Advance stages an item so that it may be retrieved via Value. Returns
+ // true iff there is an item to retrieve. Advance must be called before
+ // Value is called. May block if an item is not available.
+ Advance() bool
+ // Value returns the item that was staged by Advance. May panic if Advance
+ // returned false or was not called. Never blocks.
+ Value() ConflictInfo
+ // Err returns any error encountered by Advance. Never blocks.
+ Err() error
+ }
+ // SendStream returns the send side of the ConflictManager.StartConflictResolver client stream.
+ SendStream() interface {
+ // Send places the item onto the output stream. Returns errors
+ // encountered while sending, or if Send is called after Close or
+ // the stream has been canceled. Blocks if there is no buffer
+ // space; will unblock when buffer space is available or after
+ // the stream has been canceled.
+ Send(item ResolutionInfo) error
+ // Close indicates to the server that no more items will be sent;
+ // server Recv calls will receive io.EOF after all sent items.
+ // This is an optional call - e.g. a client might call Close if it
+ // needs to continue receiving items from the server after it's
+ // done sending. Returns errors encountered while closing, or if
+ // Close is called after the stream has been canceled. Like Send,
+ // blocks if there is no buffer space available.
+ Close() error
+ }
+}
+
+// ConflictManagerStartConflictResolverClientCall represents the call returned from ConflictManager.StartConflictResolver.
+type ConflictManagerStartConflictResolverClientCall interface {
+ ConflictManagerStartConflictResolverClientStream
+ // Finish performs the equivalent of SendStream().Close, then blocks until
+ // the server is done, and returns the positional return values for the call.
+ //
+ // Finish returns immediately if the call has been canceled; depending on the
+ // timing the output could either be an error signaling cancelation, or the
+ // valid positional return values from the server.
+ //
+ // Calling Finish is mandatory for releasing stream resources, unless the call
+ // has been canceled or any of the other methods return an error. Finish should
+ // be called at most once.
+ Finish() error
+}
+
+type implConflictManagerStartConflictResolverClientCall struct {
+ rpc.ClientCall
+ valRecv ConflictInfo
+ errRecv error
+}
+
+func (c *implConflictManagerStartConflictResolverClientCall) RecvStream() interface {
+ Advance() bool
+ Value() ConflictInfo
+ Err() error
+} {
+ return implConflictManagerStartConflictResolverClientCallRecv{c}
+}
+
+type implConflictManagerStartConflictResolverClientCallRecv struct {
+ c *implConflictManagerStartConflictResolverClientCall
+}
+
+func (c implConflictManagerStartConflictResolverClientCallRecv) Advance() bool {
+ c.c.valRecv = ConflictInfo{}
+ c.c.errRecv = c.c.Recv(&c.c.valRecv)
+ return c.c.errRecv == nil
+}
+func (c implConflictManagerStartConflictResolverClientCallRecv) Value() ConflictInfo {
+ return c.c.valRecv
+}
+func (c implConflictManagerStartConflictResolverClientCallRecv) Err() error {
+ if c.c.errRecv == io.EOF {
+ return nil
+ }
+ return c.c.errRecv
+}
+func (c *implConflictManagerStartConflictResolverClientCall) SendStream() interface {
+ Send(item ResolutionInfo) error
+ Close() error
+} {
+ return implConflictManagerStartConflictResolverClientCallSend{c}
+}
+
+type implConflictManagerStartConflictResolverClientCallSend struct {
+ c *implConflictManagerStartConflictResolverClientCall
+}
+
+func (c implConflictManagerStartConflictResolverClientCallSend) Send(item ResolutionInfo) error {
+ return c.c.Send(item)
+}
+func (c implConflictManagerStartConflictResolverClientCallSend) Close() error {
+ return c.c.CloseSend()
+}
+func (c *implConflictManagerStartConflictResolverClientCall) Finish() (err error) {
+ err = c.ClientCall.Finish()
+ return
+}
+
+// ConflictManagerServerMethods is the interface a server writer
+// implements for ConflictManager.
+//
+// ConflictManager interface provides all the methods necessary to handle
+// conflict resolution for a given database.
+type ConflictManagerServerMethods interface {
+ // StartConflictResolver registers a resolver for the database that is
+ // associated with this ConflictManager and creates a stream to receive
+ // conflicts and send resolutions.
+ // Batches of ConflictInfos will be sent over with the Continued field
+ // within the ConflictInfo representing the batch boundary. Client must
+ // respond with a batch of ResolutionInfos in the same fashion.
+ // A key is under conflict if two different values were written to it
+ // concurrently (in logical time), i.e. neither value is an ancestor of the
+ // other in the history graph.
+ // A key under conflict can be a part of a batch committed on local or
+ // remote or both syncbases. ConflictInfos for all keys in these two batches
+ // are grouped together. These keys may themselves be under conflict; the
+ // presented batch is a transitive closure of all batches containing keys
+ // under conflict.
+ // For example, for local batch {key1, key2} and remote batch {key1, key3},
+ // the batch sent for conflict resolution will be {key1, key2, key3}.
+ // If there was another concurrent batch {key2, key4}, then the batch sent
+ // for conflict resolution will be {key1, key2, key3, key4}.
+ StartConflictResolver(*context.T, ConflictManagerStartConflictResolverServerCall) error
+}
+
+// ConflictManagerServerStubMethods is the server interface containing
+// ConflictManager methods, as expected by rpc.Server.
+// The only difference between this interface and ConflictManagerServerMethods
+// is the streaming methods.
+type ConflictManagerServerStubMethods interface {
+ // StartConflictResolver registers a resolver for the database that is
+ // associated with this ConflictManager and creates a stream to receive
+ // conflicts and send resolutions.
+ // Batches of ConflictInfos will be sent over with the Continued field
+ // within the ConflictInfo representing the batch boundary. Client must
+ // respond with a batch of ResolutionInfos in the same fashion.
+ // A key is under conflict if two different values were written to it
+ // concurrently (in logical time), i.e. neither value is an ancestor of the
+ // other in the history graph.
+ // A key under conflict can be a part of a batch committed on local or
+ // remote or both syncbases. ConflictInfos for all keys in these two batches
+ // are grouped together. These keys may themselves be under conflict; the
+ // presented batch is a transitive closure of all batches containing keys
+ // under conflict.
+ // For example, for local batch {key1, key2} and remote batch {key1, key3},
+ // the batch sent for conflict resolution will be {key1, key2, key3}.
+ // If there was another concurrent batch {key2, key4}, then the batch sent
+ // for conflict resolution will be {key1, key2, key3, key4}.
+ StartConflictResolver(*context.T, *ConflictManagerStartConflictResolverServerCallStub) error
+}
+
+// ConflictManagerServerStub adds universal methods to ConflictManagerServerStubMethods.
+type ConflictManagerServerStub interface {
+ ConflictManagerServerStubMethods
+ // Describe the ConflictManager interfaces.
+ Describe__() []rpc.InterfaceDesc
+}
+
+// ConflictManagerServer returns a server stub for ConflictManager.
+// It converts an implementation of ConflictManagerServerMethods into
+// an object that may be used by rpc.Server.
+func ConflictManagerServer(impl ConflictManagerServerMethods) ConflictManagerServerStub {
+ stub := implConflictManagerServerStub{
+ impl: impl,
+ }
+ // Initialize GlobState; always check the stub itself first, to handle the
+ // case where the user has the Glob method defined in their VDL source.
+ if gs := rpc.NewGlobState(stub); gs != nil {
+ stub.gs = gs
+ } else if gs := rpc.NewGlobState(impl); gs != nil {
+ stub.gs = gs
+ }
+ return stub
+}
+
+type implConflictManagerServerStub struct {
+ impl ConflictManagerServerMethods
+ gs *rpc.GlobState
+}
+
+func (s implConflictManagerServerStub) StartConflictResolver(ctx *context.T, call *ConflictManagerStartConflictResolverServerCallStub) error {
+ return s.impl.StartConflictResolver(ctx, call)
+}
+
+func (s implConflictManagerServerStub) Globber() *rpc.GlobState {
+ return s.gs
+}
+
+func (s implConflictManagerServerStub) Describe__() []rpc.InterfaceDesc {
+ return []rpc.InterfaceDesc{ConflictManagerDesc}
+}
+
+// ConflictManagerDesc describes the ConflictManager interface.
+var ConflictManagerDesc rpc.InterfaceDesc = descConflictManager
+
+// descConflictManager hides the desc to keep godoc clean.
+var descConflictManager = rpc.InterfaceDesc{
+ Name: "ConflictManager",
+ PkgPath: "v.io/syncbase/v23/services/syncbase/nosql",
+ Doc: "// ConflictManager interface provides all the methods necessary to handle\n// conflict resolution for a given database.",
+ Methods: []rpc.MethodDesc{
+ {
+ Name: "StartConflictResolver",
+ Doc: "// StartConflictResolver registers a resolver for the database that is\n// associated with this ConflictManager and creates a stream to receive\n// conflicts and send resolutions.\n// Batches of ConflictInfos will be sent over with the Continued field\n// within the ConflictInfo representing the batch boundary. Client must\n// respond with a batch of ResolutionInfos in the same fashion.\n// A key is under conflict if two different values were written to it\n// concurrently (in logical time), i.e. neither value is an ancestor of the\n// other in the history graph.\n// A key under conflict can be a part of a batch committed on local or\n// remote or both syncbases. ConflictInfos for all keys in these two batches\n// are grouped together. These keys may themselves be under conflict; the\n// presented batch is a transitive closure of all batches containing keys\n// under conflict.\n// For example, for local batch {key1, key2} and remote batch {key1, key3},\n// the batch sent for conflict resolution will be {key1, key2, key3}.\n// If there was another concurrent batch {key2, key4}, then the batch sent\n// for conflict resolution will be {key1, key2, key3, key4}.",
+ Tags: []*vdl.Value{vdl.ValueOf(access.Tag("Write"))},
+ },
+ },
+}
+
+// ConflictManagerStartConflictResolverServerStream is the server stream for ConflictManager.StartConflictResolver.
+type ConflictManagerStartConflictResolverServerStream interface {
+ // RecvStream returns the receiver side of the ConflictManager.StartConflictResolver server stream.
+ RecvStream() interface {
+ // Advance stages an item so that it may be retrieved via Value. Returns
+ // true iff there is an item to retrieve. Advance must be called before
+ // Value is called. May block if an item is not available.
+ Advance() bool
+ // Value returns the item that was staged by Advance. May panic if Advance
+ // returned false or was not called. Never blocks.
+ Value() ResolutionInfo
+ // Err returns any error encountered by Advance. Never blocks.
+ Err() error
+ }
+ // SendStream returns the send side of the ConflictManager.StartConflictResolver server stream.
+ SendStream() interface {
+ // Send places the item onto the output stream. Returns errors encountered
+ // while sending. Blocks if there is no buffer space; will unblock when
+ // buffer space is available.
+ Send(item ConflictInfo) error
+ }
+}
+
+// ConflictManagerStartConflictResolverServerCall represents the context passed to ConflictManager.StartConflictResolver.
+type ConflictManagerStartConflictResolverServerCall interface {
+ rpc.ServerCall
+ ConflictManagerStartConflictResolverServerStream
+}
+
+// ConflictManagerStartConflictResolverServerCallStub is a wrapper that converts rpc.StreamServerCall into
+// a typesafe stub that implements ConflictManagerStartConflictResolverServerCall.
+type ConflictManagerStartConflictResolverServerCallStub struct {
+ rpc.StreamServerCall
+ valRecv ResolutionInfo
+ errRecv error
+}
+
+// Init initializes ConflictManagerStartConflictResolverServerCallStub from rpc.StreamServerCall.
+func (s *ConflictManagerStartConflictResolverServerCallStub) Init(call rpc.StreamServerCall) {
+ s.StreamServerCall = call
+}
+
+// RecvStream returns the receiver side of the ConflictManager.StartConflictResolver server stream.
+func (s *ConflictManagerStartConflictResolverServerCallStub) RecvStream() interface {
+ Advance() bool
+ Value() ResolutionInfo
+ Err() error
+} {
+ return implConflictManagerStartConflictResolverServerCallRecv{s}
+}
+
+type implConflictManagerStartConflictResolverServerCallRecv struct {
+ s *ConflictManagerStartConflictResolverServerCallStub
+}
+
+func (s implConflictManagerStartConflictResolverServerCallRecv) Advance() bool {
+ s.s.valRecv = ResolutionInfo{}
+ s.s.errRecv = s.s.Recv(&s.s.valRecv)
+ return s.s.errRecv == nil
+}
+func (s implConflictManagerStartConflictResolverServerCallRecv) Value() ResolutionInfo {
+ return s.s.valRecv
+}
+func (s implConflictManagerStartConflictResolverServerCallRecv) Err() error {
+ if s.s.errRecv == io.EOF {
+ return nil
+ }
+ return s.s.errRecv
+}
+
+// SendStream returns the send side of the ConflictManager.StartConflictResolver server stream.
+func (s *ConflictManagerStartConflictResolverServerCallStub) SendStream() interface {
+ Send(item ConflictInfo) error
+} {
+ return implConflictManagerStartConflictResolverServerCallSend{s}
+}
+
+type implConflictManagerStartConflictResolverServerCallSend struct {
+ s *ConflictManagerStartConflictResolverServerCallStub
+}
+
+func (s implConflictManagerStartConflictResolverServerCallSend) Send(item ConflictInfo) error {
+ return s.s.Send(item)
+}
+
// DatabaseClientMethods is the client interface
// containing Database methods.
//
@@ -1539,6 +1889,9 @@
// SchemaManager implements the API for managing schema metadata attached
// to a Database.
SchemaManagerClientMethods
+ // ConflictManager interface provides all the methods necessary to handle
+ // conflict resolution for a given database.
+ ConflictManagerClientMethods
// Create creates this Database.
// If perms is nil, we inherit (copy) the App perms.
// Create requires the caller to have Write permission at the App.
@@ -1579,7 +1932,7 @@
// DatabaseClient returns a client stub for Database.
func DatabaseClient(name string) DatabaseClientStub {
- return implDatabaseClientStub{name, permissions.ObjectClient(name), DatabaseWatcherClient(name), SyncGroupManagerClient(name), BlobManagerClient(name), SchemaManagerClient(name)}
+ return implDatabaseClientStub{name, permissions.ObjectClient(name), DatabaseWatcherClient(name), SyncGroupManagerClient(name), BlobManagerClient(name), SchemaManagerClient(name), ConflictManagerClient(name)}
}
type implDatabaseClientStub struct {
@@ -1590,6 +1943,7 @@
SyncGroupManagerClientStub
BlobManagerClientStub
SchemaManagerClientStub
+ ConflictManagerClientStub
}
func (c implDatabaseClientStub) Create(ctx *context.T, i0 *SchemaMetadata, i1 access.Permissions, opts ...rpc.CallOpt) (err error) {
@@ -1786,6 +2140,9 @@
// SchemaManager implements the API for managing schema metadata attached
// to a Database.
SchemaManagerServerMethods
+ // ConflictManager interface provides all the methods necessary to handle
+ // conflict resolution for a given database.
+ ConflictManagerServerMethods
// Create creates this Database.
// If perms is nil, we inherit (copy) the App perms.
// Create requires the caller to have Write permission at the App.
@@ -1899,6 +2256,9 @@
// SchemaManager implements the API for managing schema metadata attached
// to a Database.
SchemaManagerServerStubMethods
+ // ConflictManager interface provides all the methods necessary to handle
+ // conflict resolution for a given database.
+ ConflictManagerServerStubMethods
// Create creates this Database.
// If perms is nil, we inherit (copy) the App perms.
// Create requires the caller to have Write permission at the App.
@@ -1949,6 +2309,7 @@
SyncGroupManagerServerStub: SyncGroupManagerServer(impl),
BlobManagerServerStub: BlobManagerServer(impl),
SchemaManagerServerStub: SchemaManagerServer(impl),
+ ConflictManagerServerStub: ConflictManagerServer(impl),
}
// Initialize GlobState; always check the stub itself first, to handle the
// case where the user has the Glob method defined in their VDL source.
@@ -1967,6 +2328,7 @@
SyncGroupManagerServerStub
BlobManagerServerStub
SchemaManagerServerStub
+ ConflictManagerServerStub
gs *rpc.GlobState
}
@@ -2003,7 +2365,7 @@
}
func (s implDatabaseServerStub) Describe__() []rpc.InterfaceDesc {
- return []rpc.InterfaceDesc{DatabaseDesc, permissions.ObjectDesc, DatabaseWatcherDesc, watch.GlobWatcherDesc, SyncGroupManagerDesc, BlobManagerDesc, SchemaManagerDesc}
+ return []rpc.InterfaceDesc{DatabaseDesc, permissions.ObjectDesc, DatabaseWatcherDesc, watch.GlobWatcherDesc, SyncGroupManagerDesc, BlobManagerDesc, SchemaManagerDesc, ConflictManagerDesc}
}
// DatabaseDesc describes the Database interface.
@@ -2020,6 +2382,7 @@
{"SyncGroupManager", "v.io/syncbase/v23/services/syncbase/nosql", "// SyncGroupManager is the interface for SyncGroup operations.\n// TODO(hpucha): Add blessings to create/join and add a refresh method."},
{"BlobManager", "v.io/syncbase/v23/services/syncbase/nosql", "// BlobManager is the interface for blob operations."},
{"SchemaManager", "v.io/syncbase/v23/services/syncbase/nosql", "// SchemaManager implements the API for managing schema metadata attached\n// to a Database."},
+ {"ConflictManager", "v.io/syncbase/v23/services/syncbase/nosql", "// ConflictManager interface provides all the methods necessary to handle\n// conflict resolution for a given database."},
},
Methods: []rpc.MethodDesc{
{
diff --git a/v23/services/syncbase/nosql/types.vdl b/v23/services/syncbase/nosql/types.vdl
index eae8e9d..407e956 100644
--- a/v23/services/syncbase/nosql/types.vdl
+++ b/v23/services/syncbase/nosql/types.vdl
@@ -111,6 +111,158 @@
Defer
}
+// ConflictInfo contains information to fully specify a conflict
+// for a key, providing the (local, remote, ancestor) tuple.
+// A key under conflict can be a part of a batch in local, remote or both
+// updates. Since the batches can have more than one key, all ConflictInfos
+// for the keys within the batches are grouped together into a single conflict
+// batch and sent as a stream with the Continued field representing conflict
+// batch boundaries.
+type ConflictInfo struct {
+ // Data is a unit chunk of ConflictInfo which can be sent over the conflict
+ // stream.
+ Data ConflictData
+
+ // Continued represents whether the batch of ConflictInfos has ended.
+ Continued bool
+}
+
+// ConflictData represents a unit of conflict data sent over the stream. It
+// can either contain information about a Batch or about an operation done
+// on a row.
+type ConflictData union {
+ Batch BatchInfo
+ Row RowInfo
+}
+
+type BatchInfo struct {
+ // Id is an identifier for a batch contained in a conflict. It is
+ // unique only in the context of a given conflict. Its purpose is solely to
+ // group one or more RowInfo objects together to represent a batch that
+ // was committed by the client.
+ Id uint16
+
+ // Hint is the hint provided by the client when this batch was committed.
+ Hint string
+
+ // Source states where the batch comes from.
+ Source BatchSource
+}
+
+// BatchSource represents where the batch was committed.
+type BatchSource enum {
+ // Local represents that the batch was committed on local syncbase.
+ Local
+
+ // Remote represents that the batch was committed on remote syncbase.
+ Remote
+}
+
+// RowInfo contains a single operation performed on a row (in case of read or
+// write) or a range or rows (in case of scan) along with a mapping to each
+// of the batches that this operation belongs to.
+// For example, if Row1 was updated on local syncbase conflicting with a write
+// on remote syncbase as part of two separate batches, then it will be
+// represented by a single RowInfo with Write Operation containing the
+// respective local and remote values along with the batch id for both batches
+// stored in the BatchIds field.
+type RowInfo struct {
+ // Op is a specific operation represented by RowInfo
+ Op Operation
+ // BatchIds contains ids of all batches that this RowInfo is a part of.
+ BatchIds []uint16
+}
+
+// Operation represents a specific operation on a row or a set of rows that is
+// a part of the conflict.
+type Operation union {
+ // Read represents a read operation performed on a specific row. For a given
+ // row key there can only be at max one Read operation within a conflict.
+ Read RowOp
+
+ // Write represents a write operation performed on a specific row. For a
+ // given row key there can only be at max one Write operation within a
+ // conflict.
+ Write RowOp
+
+ // Scan represents a scan operation performed over a specific range of keys.
+ // For a given key range there can be at max one ScanOp within the Conflict.
+ Scan ScanOp
+}
+
+// RowOp represents a read or write operation on a row corresponding to the
+// given key.
+type RowOp struct {
+ // The key under conflict.
+ Key string
+
+ // LocalValue contains the value read or written by local syncbase or nil.
+ LocalValue ?Value
+
+ // RemoteValue contains the value read or written by remote syncbase or nil.
+ RemoteValue ?Value
+
+ // AncestorValue contains the value for the key which is the lowest common
+ // ancestor of the two values represented by LocalValue and RemoteValue or
+ // nil if no ancestor exists or if the operation was read.
+ AncestorValue ?Value
+}
+
+// ScanOp provides details of a scan operation.
+type ScanOp struct {
+ // Start contains the starting key for a range scan.
+ Start string
+
+ // Limit contains the end key for a range scan.
+ Limit string
+}
+
+// Value contains the encoded bytes for a row's value stored in syncbase.
+type Value struct {
+ // VOM encoded bytes for a row's value
+ Bytes []byte
+
+ // Write timestamp for this value in nanoseconds.
+ // TODO(jlodhia): change the timestamp to vdl.time here, in commit timestamp
+ // and clock data.
+ WriteTs int64
+}
+
+// ValueSelection represents the value that was selected as the final resolution
+// for a conflict.
+type ValueSelection enum {
+ // Local should be used if local value is picked as the final result.
+ Local
+
+ // Remote should be used if remote value is picked as the final result.
+ Remote
+
+ // Other should be used if a new value different from local and remote is
+ // used as the final result.
+ Other
+}
+
+// ResolutionInfo contains the application’s reply to a conflict for a key,
+// providing the resolution value. The resolution may be over a group of keys
+// in which case the application must send a stream of ResolutionInfos with
+// the Continued field for the last ResolutionInfo representing the end of the
+// batch with a value false. ResolutionInfos sent as part of a batch will be
+// committed as a batch. If the commit fails, the Conflict will be re-sent.
+type ResolutionInfo struct {
+ // Key is the key under conflict.
+ Key string
+
+ // Selection represents the value that was selected as resolution.
+ Selection ValueSelection
+
+ // Result is the resolved value for the key. This field should be used only
+ // if value of Selection field is 'Other'
+ Result ?Value
+
+ // Continued represents whether the batch of ResolutionInfos has ended.
+ Continued bool
+}
+
// SchemaMetadata maintains metadata related to the schema of a given database.
// There is one SchemaMetadata per database.
type SchemaMetadata struct {
diff --git a/v23/services/syncbase/nosql/types.vdl.go b/v23/services/syncbase/nosql/types.vdl.go
index adb4e66..6a8e0c1 100644
--- a/v23/services/syncbase/nosql/types.vdl.go
+++ b/v23/services/syncbase/nosql/types.vdl.go
@@ -166,6 +166,336 @@
}) {
}
+// ConflictInfo contains information to fully specify a conflict
+// for a key, providing the (local, remote, ancestor) tuple.
+// A key under conflict can be a part of a batch in local, remote or both
+// updates. Since the batches can have more than one key, all ConflictInfos
+// for the keys within the batches are grouped together into a single conflict
+// batch and sent as a stream with the Continued field representing conflict
+// batch boundaries.
+type ConflictInfo struct {
+ // Data is a unit chunk of ConflictInfo which can be sent over the conflict
+ // stream.
+ Data ConflictData
+ // Continued represents whether the batch of ConflictInfos has ended.
+ Continued bool
+}
+
+func (ConflictInfo) __VDLReflect(struct {
+ Name string `vdl:"v.io/syncbase/v23/services/syncbase/nosql.ConflictInfo"`
+}) {
+}
+
+type (
+ // ConflictData represents any single field of the ConflictData union type.
+ //
+ // ConflictData represents a unit of conflict data sent over the stream. It
+ // can either contain information about a Batch or about an operation done
+ // on a row.
+ ConflictData interface {
+ // Index returns the field index.
+ Index() int
+ // Interface returns the field value as an interface.
+ Interface() interface{}
+ // Name returns the field name.
+ Name() string
+ // __VDLReflect describes the ConflictData union type.
+ __VDLReflect(__ConflictDataReflect)
+ }
+ // ConflictDataBatch represents field Batch of the ConflictData union type.
+ ConflictDataBatch struct{ Value BatchInfo }
+ // ConflictDataRow represents field Row of the ConflictData union type.
+ ConflictDataRow struct{ Value RowInfo }
+ // __ConflictDataReflect describes the ConflictData union type.
+ __ConflictDataReflect struct {
+ Name string `vdl:"v.io/syncbase/v23/services/syncbase/nosql.ConflictData"`
+ Type ConflictData
+ Union struct {
+ Batch ConflictDataBatch
+ Row ConflictDataRow
+ }
+ }
+)
+
+func (x ConflictDataBatch) Index() int { return 0 }
+func (x ConflictDataBatch) Interface() interface{} { return x.Value }
+func (x ConflictDataBatch) Name() string { return "Batch" }
+func (x ConflictDataBatch) __VDLReflect(__ConflictDataReflect) {}
+
+func (x ConflictDataRow) Index() int { return 1 }
+func (x ConflictDataRow) Interface() interface{} { return x.Value }
+func (x ConflictDataRow) Name() string { return "Row" }
+func (x ConflictDataRow) __VDLReflect(__ConflictDataReflect) {}
+
+type BatchInfo struct {
+ // Id is an identifier for a batch contained in a conflict. It is
+ // unique only in the context of a given conflict. Its purpose is solely to
+ // group one or more RowInfo objects together to represent a batch that
+ // was committed by the client.
+ Id uint16
+ // Hint is the hint provided by the client when this batch was committed.
+ Hint string
+ // Source states where the batch comes from.
+ Source BatchSource
+}
+
+func (BatchInfo) __VDLReflect(struct {
+ Name string `vdl:"v.io/syncbase/v23/services/syncbase/nosql.BatchInfo"`
+}) {
+}
+
+// BatchSource represents where the batch was committed.
+type BatchSource int
+
+const (
+ BatchSourceLocal BatchSource = iota
+ BatchSourceRemote
+)
+
+// BatchSourceAll holds all labels for BatchSource.
+var BatchSourceAll = [...]BatchSource{BatchSourceLocal, BatchSourceRemote}
+
+// BatchSourceFromString creates a BatchSource from a string label.
+func BatchSourceFromString(label string) (x BatchSource, err error) {
+ err = x.Set(label)
+ return
+}
+
+// Set assigns label to x.
+func (x *BatchSource) Set(label string) error {
+ switch label {
+ case "Local", "local":
+ *x = BatchSourceLocal
+ return nil
+ case "Remote", "remote":
+ *x = BatchSourceRemote
+ return nil
+ }
+ *x = -1
+ return fmt.Errorf("unknown label %q in nosql.BatchSource", label)
+}
+
+// String returns the string label of x.
+func (x BatchSource) String() string {
+ switch x {
+ case BatchSourceLocal:
+ return "Local"
+ case BatchSourceRemote:
+ return "Remote"
+ }
+ return ""
+}
+
+func (BatchSource) __VDLReflect(struct {
+ Name string `vdl:"v.io/syncbase/v23/services/syncbase/nosql.BatchSource"`
+ Enum struct{ Local, Remote string }
+}) {
+}
+
+// RowInfo contains a single operation performed on a row (in case of read or
+// write) or a range or rows (in case of scan) along with a mapping to each
+// of the batches that this operation belongs to.
+// For example, if Row1 was updated on local syncbase conflicting with a write
+// on remote syncbase as part of two separate batches, then it will be
+// represented by a single RowInfo with Write Operation containing the
+// respective local and remote values along with the batch id for both batches
+// stored in the BatchIds field.
+type RowInfo struct {
+ // Op is a specific operation represented by RowInfo
+ Op Operation
+ // BatchIds contains ids of all batches that this RowInfo is a part of.
+ BatchIds []uint16
+}
+
+func (RowInfo) __VDLReflect(struct {
+ Name string `vdl:"v.io/syncbase/v23/services/syncbase/nosql.RowInfo"`
+}) {
+}
+
+type (
+ // Operation represents any single field of the Operation union type.
+ //
+ // Operation represents a specific operation on a row or a set of rows that is
+ // a part of the conflict.
+ Operation interface {
+ // Index returns the field index.
+ Index() int
+ // Interface returns the field value as an interface.
+ Interface() interface{}
+ // Name returns the field name.
+ Name() string
+ // __VDLReflect describes the Operation union type.
+ __VDLReflect(__OperationReflect)
+ }
+ // OperationRead represents field Read of the Operation union type.
+ //
+ // Read represents a read operation performed on a specific row. For a given
+ // row key there can only be at max one Read operation within a conflict.
+ OperationRead struct{ Value RowOp }
+ // OperationWrite represents field Write of the Operation union type.
+ //
+ // Write represents a write operation performed on a specific row. For a
+ // given row key there can only be at max one Write operation within a
+ // conflict.
+ OperationWrite struct{ Value RowOp }
+ // OperationScan represents field Scan of the Operation union type.
+ //
+ // Scan represents a scan operation performed over a specific range of keys.
+ // For a given key range there can be at max one ScanOp within the Conflict.
+ OperationScan struct{ Value ScanOp }
+ // __OperationReflect describes the Operation union type.
+ __OperationReflect struct {
+ Name string `vdl:"v.io/syncbase/v23/services/syncbase/nosql.Operation"`
+ Type Operation
+ Union struct {
+ Read OperationRead
+ Write OperationWrite
+ Scan OperationScan
+ }
+ }
+)
+
+func (x OperationRead) Index() int { return 0 }
+func (x OperationRead) Interface() interface{} { return x.Value }
+func (x OperationRead) Name() string { return "Read" }
+func (x OperationRead) __VDLReflect(__OperationReflect) {}
+
+func (x OperationWrite) Index() int { return 1 }
+func (x OperationWrite) Interface() interface{} { return x.Value }
+func (x OperationWrite) Name() string { return "Write" }
+func (x OperationWrite) __VDLReflect(__OperationReflect) {}
+
+func (x OperationScan) Index() int { return 2 }
+func (x OperationScan) Interface() interface{} { return x.Value }
+func (x OperationScan) Name() string { return "Scan" }
+func (x OperationScan) __VDLReflect(__OperationReflect) {}
+
+// RowOp represents a read or write operation on a row corresponding to the
+// given key.
+type RowOp struct {
+ // The key under conflict.
+ Key string
+ // LocalValue contains the value read or written by local syncbase or nil.
+ LocalValue *Value
+ // RemoteValue contains the value read or written by remote syncbase or nil.
+ RemoteValue *Value
+ // AncestorValue contains the value for the key which is the lowest common
+ // ancestor of the two values represented by LocalValue and RemoteValue or
+ // nil if no ancestor exists or if the operation was read.
+ AncestorValue *Value
+}
+
+func (RowOp) __VDLReflect(struct {
+ Name string `vdl:"v.io/syncbase/v23/services/syncbase/nosql.RowOp"`
+}) {
+}
+
+// ScanOp provides details of a scan operation.
+type ScanOp struct {
+ // Start contains the starting key for a range scan.
+ Start string
+ // Limit contains the end key for a range scan.
+ Limit string
+}
+
+func (ScanOp) __VDLReflect(struct {
+ Name string `vdl:"v.io/syncbase/v23/services/syncbase/nosql.ScanOp"`
+}) {
+}
+
+// Value contains the encoded bytes for a row's value stored in syncbase.
+type Value struct {
+ // VOM encoded bytes for a row's value
+ Bytes []byte
+ // Write timestamp for this value in nanoseconds.
+ // TODO(jlodhia): change the timestamp to vdl.time here, in commit timestamp
+ // and clock data.
+ WriteTs int64
+}
+
+func (Value) __VDLReflect(struct {
+ Name string `vdl:"v.io/syncbase/v23/services/syncbase/nosql.Value"`
+}) {
+}
+
+// ValueSelection represents the value that was selected as the final resolution
+// for a conflict.
+type ValueSelection int
+
+const (
+ ValueSelectionLocal ValueSelection = iota
+ ValueSelectionRemote
+ ValueSelectionOther
+)
+
+// ValueSelectionAll holds all labels for ValueSelection.
+var ValueSelectionAll = [...]ValueSelection{ValueSelectionLocal, ValueSelectionRemote, ValueSelectionOther}
+
+// ValueSelectionFromString creates a ValueSelection from a string label.
+func ValueSelectionFromString(label string) (x ValueSelection, err error) {
+ err = x.Set(label)
+ return
+}
+
+// Set assigns label to x.
+func (x *ValueSelection) Set(label string) error {
+ switch label {
+ case "Local", "local":
+ *x = ValueSelectionLocal
+ return nil
+ case "Remote", "remote":
+ *x = ValueSelectionRemote
+ return nil
+ case "Other", "other":
+ *x = ValueSelectionOther
+ return nil
+ }
+ *x = -1
+ return fmt.Errorf("unknown label %q in nosql.ValueSelection", label)
+}
+
+// String returns the string label of x.
+func (x ValueSelection) String() string {
+ switch x {
+ case ValueSelectionLocal:
+ return "Local"
+ case ValueSelectionRemote:
+ return "Remote"
+ case ValueSelectionOther:
+ return "Other"
+ }
+ return ""
+}
+
+func (ValueSelection) __VDLReflect(struct {
+ Name string `vdl:"v.io/syncbase/v23/services/syncbase/nosql.ValueSelection"`
+ Enum struct{ Local, Remote, Other string }
+}) {
+}
+
+// ResolutionInfo contains the application’s reply to a conflict for a key,
+// providing the resolution value. The resolution may be over a group of keys
+// in which case the application must send a stream of ResolutionInfos with
+// the Continued field for the last ResolutionInfo representing the end of the
+// batch with a value false. ResolutionInfos sent as part of a batch will be
+// committed as a batch. If the commit fails, the Conflict will be re-sent.
+type ResolutionInfo struct {
+ // Key is the key under conflict.
+ Key string
+ // Selection represents the value that was selected as resolution.
+ Selection ValueSelection
+ // Result is the resolved value for the key. This field should be used only
+ // if value of Selection field is 'Other'
+ Result *Value
+ // Continued represents whether the batch of ResolutionInfos has ended.
+ Continued bool
+}
+
+func (ResolutionInfo) __VDLReflect(struct {
+ Name string `vdl:"v.io/syncbase/v23/services/syncbase/nosql.ResolutionInfo"`
+}) {
+}
+
// SchemaMetadata maintains metadata related to the schema of a given database.
// There is one SchemaMetadata per database.
type SchemaMetadata struct {
@@ -319,6 +649,17 @@
vdl.Register((*SyncGroupSpec)(nil))
vdl.Register((*SyncGroupMemberInfo)(nil))
vdl.Register((*ResolverType)(nil))
+ vdl.Register((*ConflictInfo)(nil))
+ vdl.Register((*ConflictData)(nil))
+ vdl.Register((*BatchInfo)(nil))
+ vdl.Register((*BatchSource)(nil))
+ vdl.Register((*RowInfo)(nil))
+ vdl.Register((*Operation)(nil))
+ vdl.Register((*RowOp)(nil))
+ vdl.Register((*ScanOp)(nil))
+ vdl.Register((*Value)(nil))
+ vdl.Register((*ValueSelection)(nil))
+ vdl.Register((*ResolutionInfo)(nil))
vdl.Register((*SchemaMetadata)(nil))
vdl.Register((*CrPolicy)(nil))
vdl.Register((*CrRule)(nil))
diff --git a/v23/syncbase/nosql/cr_connection_test.go b/v23/syncbase/nosql/cr_connection_test.go
new file mode 100644
index 0000000..b99c15d
--- /dev/null
+++ b/v23/syncbase/nosql/cr_connection_test.go
@@ -0,0 +1,138 @@
+// 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 nosql
+
+import (
+ "fmt"
+ "testing"
+ "time"
+
+ wire "v.io/syncbase/v23/services/syncbase/nosql"
+ "v.io/syncbase/v23/syncbase/nosql/crtestutil"
+ "v.io/v23/context"
+)
+
+type resolverImpl1 struct {
+}
+
+func (ri *resolverImpl1) OnConflict(ctx *context.T, conflict *Conflict) (r Resolution) {
+ return
+}
+
+// TestCrConnectionClose tests if db.Close() correctly shuts down the goroutine
+// that runs conflict resolution code in an infinite loop.
+// Setup:
+// - A mutex lock is used to cause Advance() on the receive stream to block.
+// This lock is held by the test before starting off the CR thread.
+// - ctx.Cancel() is simulated by releasing the mutex lock. This is done right
+// after calling db.Close()
+// - The mutex is held again by the test to cause the CR thread to block on
+// Advance() in case the CR thead did not shutdown. This will lead to
+// failure of the test.
+// Expected result:
+// - The CR thread should not get/remain blocked on Advance() after db.Close()
+func TestCrConnectionClose(t *testing.T) {
+ db := NewDatabase("parentName", "db1", getSchema(&resolverImpl1{}))
+ advance := func(st *crtestutil.State) bool {
+ fmt.Println("Advance() for ConflictStreamImpl called")
+ st.IsBlocked = true
+ st.Mu.Lock()
+ defer st.Mu.Unlock()
+ fmt.Println("Advance() for ConflictStreamImpl returning")
+ st.IsBlocked = false
+ return false
+ }
+
+ st := &crtestutil.State{}
+ crStream := &crtestutil.CrStreamImpl{
+ C: &crtestutil.ConflictStreamImpl{St: st, AdvanceFn: advance},
+ R: &crtestutil.ResolutionStreamImpl{St: st},
+ }
+ db.c = crtestutil.MockDbClient(db.c, crStream)
+ db.crState.reconnectWaitTime = 10 * time.Millisecond
+
+ ctx, _ := context.RootContext()
+ st.Mu.Lock() // causes Advance() to block
+ db.EnforceSchema(ctx) // kicks off the CR thread
+ for i := 0; i < 100 && !st.IsBlocked; i++ { // wait till Advance() call is blocked
+ time.Sleep(time.Millisecond)
+ }
+ db.Close()
+ st.Mu.Unlock() // simulate ctx cancel()
+ for i := 0; i < 100 && st.IsBlocked; i++ {
+ time.Sleep(time.Millisecond)
+ }
+
+ st.Mu.Lock() // causes Advance() to block if called again
+ // let conflict resolution routine complete its cleanup.
+ // If the code does not work as intended then it will end up reconnecting
+ // the stream and blocking on the mutex
+ time.Sleep(time.Millisecond)
+ if st.IsBlocked {
+ t.Error("Error: The conflict resolution routine did not die")
+ }
+}
+
+// TestCrConnectionReestablish tests if the CR thread reestablishes a broken
+// connection stream with syncbase.
+// Setup:
+// - Advance() is allowed to return immediately 3 times with return value
+// false signifying 3 consecutive breaks in the connection.
+// - A mutex lock is used to cause Advance() on the recieve stream to block
+// on the 4th attempt. This lock is held by the test before starting off the
+// CR thread. This simulates a stable connection between CR and syncbase.
+// Expected result:
+// - Advance() should get blocked after some time.
+// - st.AdvanceCount should be equal to 3
+func TestCrConnectionReestablish(t *testing.T) {
+ db := NewDatabase("parentName", "db1", getSchema(&resolverImpl1{}))
+ advance := func(st *crtestutil.State) bool {
+ fmt.Println("Advance() for ConflictStreamImpl called")
+ st.AdvanceCount++
+ if st.AdvanceCount > 2 {
+ // Connection stays stable after 3 attempts
+ st.IsBlocked = true
+ st.Mu.Lock()
+ defer st.Mu.Unlock()
+ }
+ fmt.Println("Advance() for ConflictStreamImpl returning")
+ st.IsBlocked = false
+ return false
+ }
+
+ st := &crtestutil.State{}
+ crStream := &crtestutil.CrStreamImpl{
+ C: &crtestutil.ConflictStreamImpl{St: st, AdvanceFn: advance},
+ R: &crtestutil.ResolutionStreamImpl{St: st},
+ }
+ db.c = crtestutil.MockDbClient(db.c, crStream)
+ db.crState.reconnectWaitTime = 10 * time.Millisecond
+
+ ctx, _ := context.RootContext()
+ st.Mu.Lock() // causes Advance() to block
+ db.EnforceSchema(ctx)
+ for i := 0; i < 100 && !st.IsBlocked; i++ {
+ time.Sleep(time.Millisecond) // wait till Advance() call is blocked
+ }
+
+ if !st.IsBlocked {
+ t.Error("Error: Advance() did not block")
+ }
+ if st.AdvanceCount != 3 {
+ t.Errorf("Error: Advance() expected to be called 3 times, instead called %d times", st.AdvanceCount)
+ }
+
+ // Shutdown the cr routine
+ db.Close()
+ st.Mu.Unlock()
+}
+
+func getSchema(cr ConflictResolver) *Schema {
+ return &Schema{
+ Metadata: wire.SchemaMetadata{},
+ Upgrader: nil,
+ Resolver: cr,
+ }
+}
diff --git a/v23/syncbase/nosql/cr_test.go b/v23/syncbase/nosql/cr_test.go
new file mode 100644
index 0000000..f9618d9
--- /dev/null
+++ b/v23/syncbase/nosql/cr_test.go
@@ -0,0 +1,605 @@
+// 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 nosql
+
+import (
+ "bytes"
+ "strings"
+ "testing"
+ "time"
+
+ wire "v.io/syncbase/v23/services/syncbase/nosql"
+ "v.io/syncbase/v23/syncbase/nosql/crtestutil"
+ "v.io/v23/context"
+ "v.io/v23/vom"
+)
+
+// TODO(jlodhia): extract this test (along with the crtest helpers) to
+// nosql_test, or alternatively, move the crtest helper to an internal dir.
+
+
+////////////////////////////////////////
+// Client Code
+
+// Client constants
+const (
+ TaskIdPrefix = "task:"
+ ListIdPrefix = "list:"
+
+ List1 = "list:1"
+
+ Task1 = "task:1:1"
+ Task2 = "task:1:2"
+
+ HintListDelete = "ListDelete"
+
+ HintTextEdit = "TaskTextEdit"
+ HintDone = "TaskDoneToggle"
+ HintTaskDelete = "TaskDelete"
+ HintTaskAdd = "TaskAdd"
+
+ TextNew = "EditedText"
+)
+
+// Client data structures
+type Task struct {
+ Text string
+ Done bool
+ DoneTs int64
+ ListId string
+}
+
+type List struct {
+ Name string
+ TaskCount int16
+ DoneCount int16
+}
+
+// Client conflict reolution impl
+type ConflictResolverImpl struct {
+}
+
+func (ri *ConflictResolverImpl) OnConflict(ctx *context.T, conflict *Conflict) Resolution {
+ taskRows := findRowsByType(conflict, TaskIdPrefix)
+ result := ri.handleTasks(ctx, taskRows)
+
+ listRows := findRowsByType(conflict, ListIdPrefix)
+ ri.handleLists(ctx, listRows, result)
+ return Resolution{ResultSet: result}
+}
+
+// handleTasks does a merge of fields within each task under conflict.
+// It handles conflict on fields as follows:
+// Text: Last writer wins
+// Done: Depends on DoneTs
+// DoneTs: The side that is not equal to ancestor and is the lower of the two
+// wins. This means that if a user marks a task done on one device
+// and then later marks the same task done on another device that has
+// not synced yet, the done timestamp that is selected will be from the
+// former action.
+func (ri *ConflictResolverImpl) handleTasks(ctx *context.T, tasks []ConflictRow) map[string]ResolvedRow {
+ result := map[string]ResolvedRow{}
+ for _, t := range tasks {
+ if isRowAdded(t) {
+ result[t.Key] = handleRowAdd(t)
+ continue
+ }
+ if isRowDeleted(t) {
+ result[t.Key] = ResolvedRow{Key: t.Key, Result: nil}
+ continue
+ }
+
+ var localTask, remoteTask, ancestorTask, resolvedTask Task
+ t.LocalValue.Get(&localTask)
+ t.RemoteValue.Get(&remoteTask)
+ t.AncestorValue.Get(&ancestorTask)
+
+ // prime the reolved task with local version for simplicity.
+ resolvedTask = localTask
+
+ // check field text for conflict
+ if localTask.Text != remoteTask.Text {
+ if remoteTask.Text != ancestorTask.Text {
+ if (localTask.Text == ancestorTask.Text) || (t.RemoteValue.WriteTs.After(t.LocalValue.WriteTs)) {
+ resolvedTask.Text = remoteTask.Text
+ }
+ }
+ }
+
+ // check field done for conflict
+ if localTask.DoneTs != remoteTask.DoneTs {
+ if remoteTask.DoneTs != ancestorTask.DoneTs {
+ if (localTask.DoneTs == ancestorTask.DoneTs) || (remoteTask.DoneTs < localTask.DoneTs) {
+ resolvedTask.DoneTs = remoteTask.DoneTs
+ resolvedTask.Done = remoteTask.Done
+ }
+ }
+ }
+ resValue, _ := NewValue(ctx, &resolvedTask)
+ result[t.Key] = ResolvedRow{Key: t.Key, Result: resValue}
+ }
+ return result
+}
+
+// handleLists does a merge of fields within each list under conflict. If a list
+// is deleted, it looks at the resolution map, finds all tasks that belong to
+// the list and changes the resolution to delete for those tasks.
+// It handles conflict on fields as follows:
+// Name: Last writer wins
+// TaskCount: Add both counts and subtract the ancestor count.
+// DoneCount: Add both counts and subtract the ancestor count.
+// NOTE: The above counters are resolved in a naive way for simplicity of this
+// test. There are many other ways to represent the counters in a more correct
+// way.
+func (ri *ConflictResolverImpl) handleLists(ctx *context.T, lists []ConflictRow, resolution map[string]ResolvedRow) {
+ for _, l := range lists {
+ if isRowAdded(l) {
+ resolution[l.Key] = handleRowAdd(l)
+ continue
+ }
+ if isRowDeleted(l) {
+ resolution[l.Key] = ResolvedRow{Key: l.Key, Result: nil}
+ deleteTasksForList(resolution, l.Key)
+ continue
+ }
+
+ var localList, remoteList, ancestorList, resolvedList List
+ l.LocalValue.Get(&localList)
+ l.RemoteValue.Get(&remoteList)
+ l.AncestorValue.Get(&ancestorList)
+
+ // prime the reolved task with local version for simplicity.
+ resolvedList = localList
+
+ // check field Name for conflict
+ if localList.Name != remoteList.Name {
+ if remoteList.Name != ancestorList.Name {
+ if (localList.Name == ancestorList.Name) || (l.RemoteValue.WriteTs.After(l.LocalValue.WriteTs)) {
+ resolvedList.Name = remoteList.Name
+ }
+ }
+ }
+
+ // check field TaskCount for conflict
+ if localList.TaskCount != remoteList.TaskCount {
+ if remoteList.TaskCount != ancestorList.TaskCount {
+ if localList.TaskCount == ancestorList.TaskCount {
+ resolvedList.TaskCount = remoteList.TaskCount
+ } else {
+ resolvedList.TaskCount = localList.TaskCount + remoteList.TaskCount - ancestorList.TaskCount
+ }
+ }
+ }
+
+ // check field DoneCount for conflict
+ if localList.DoneCount != remoteList.DoneCount {
+ if remoteList.DoneCount != ancestorList.DoneCount {
+ if localList.DoneCount == ancestorList.DoneCount {
+ resolvedList.DoneCount = remoteList.DoneCount
+ } else {
+ resolvedList.DoneCount = localList.DoneCount + remoteList.DoneCount - ancestorList.DoneCount
+ }
+ }
+ }
+ resValue, _ := NewValue(ctx, &resolvedList)
+ resolution[l.Key] = ResolvedRow{Key: l.Key, Result: resValue}
+ }
+}
+
+func deleteTasksForList(resolution map[string]ResolvedRow, listKey string) {
+ // find key prefix for tasks belonging to list
+ taskKeyPrefix := strings.Replace(listKey, ListIdPrefix, TaskIdPrefix, 1) + ":"
+ for k, v := range resolution {
+ if strings.HasPrefix(k, taskKeyPrefix) {
+ v.Result = nil
+ resolution[k] = v
+ }
+ }
+}
+
+func handleRowAdd(r ConflictRow) ResolvedRow {
+ var resolvedTask *Value
+ // first one to add wins
+ if r.RemoteValue == nil {
+ resolvedTask = r.LocalValue
+ } else if r.LocalValue == nil {
+ resolvedTask = r.RemoteValue
+ } else if r.LocalValue.WriteTs.Before(r.RemoteValue.WriteTs) {
+ resolvedTask = r.LocalValue
+ } else {
+ resolvedTask = r.RemoteValue
+ }
+ return ResolvedRow{Key: r.Key, Result: resolvedTask}
+}
+
+func isRowDeleted(r ConflictRow) bool {
+ return r.AncestorValue != nil &&
+ (r.LocalValue == nil || r.RemoteValue == nil)
+}
+
+func isRowAdded(r ConflictRow) bool {
+ return r.AncestorValue == nil &&
+ (r.LocalValue != nil || r.RemoteValue != nil)
+}
+
+func findRowsByType(conflict *Conflict, typePrefix string) []ConflictRow {
+ rows := []ConflictRow{}
+ if conflict.WriteSet != nil {
+ for k, v := range conflict.WriteSet.ByKey {
+ if strings.HasPrefix(k, typePrefix) {
+ rows = append(rows, v)
+ }
+ }
+ }
+ return rows
+}
+
+////////////////////////////////////////
+// Tests
+
+func TestConflictResolutionSingleObject(t *testing.T) {
+ mockConflit, expResult := simpleConflictStream(t)
+ RunTest(t, mockConflit, expResult, false)
+}
+
+func TestConflictResolutionAddDelete(t *testing.T) {
+ mockConflit, expResult := addDeleteConflictStream(t)
+ RunTest(t, mockConflit, expResult, true)
+}
+
+func TestConflictResolutionIntersectingConflicts(t *testing.T) {
+ mockConflict, expResult := twoIntersectingBatchesConflictStream(t)
+ RunTest(t, mockConflict, expResult, true)
+}
+
+func TestConflictResolutionListDelete(t *testing.T) {
+ mockConflit, expResult := listDeleteConflictsWithTaskAdd(t)
+ RunTest(t, mockConflit, expResult, true)
+}
+
+func RunTest(t *testing.T, mockConflit []wire.ConflictInfo, expResult map[string]wire.ResolutionInfo, singleBatch bool) {
+ db := NewDatabase("parentName", "db1", getSchema(&ConflictResolverImpl{}))
+ advance := func(st *crtestutil.State) bool {
+ if st.ValIndex >= len(mockConflit) {
+ st.IsBlocked = true
+ st.Mu.Lock()
+ defer st.Mu.Unlock()
+ }
+ st.IsBlocked = false
+ st.Val = mockConflit[st.ValIndex]
+ st.ValIndex++
+ return true
+ }
+
+ st := &crtestutil.State{}
+ crStream := &crtestutil.CrStreamImpl{
+ C: &crtestutil.ConflictStreamImpl{St: st, AdvanceFn: advance},
+ R: &crtestutil.ResolutionStreamImpl{St: st},
+ }
+ db.c = crtestutil.MockDbClient(db.c, crStream)
+ db.crState.reconnectWaitTime = 10 * time.Millisecond
+
+ ctx, _ := context.RootContext()
+ st.Mu.Lock() // causes Advance() to block
+ db.EnforceSchema(ctx)
+ for i := 0; i < 100 && (len(st.Result) != len(expResult)); i++ {
+ time.Sleep(time.Millisecond) // wait till Advance() call is blocked
+ }
+ if len(st.Result) != len(expResult) {
+ t.Errorf("\n Unexpected number of results. Expected: %d, actual: %d", len(expResult), len(st.Result))
+ }
+ for i, result := range st.Result {
+ compareResult(t, expResult[result.Key], result)
+ if singleBatch {
+ if result.Continued != (i < len(st.Result)-1) {
+ t.Error("\nUnexpected value for continued in single batch")
+ }
+ } else {
+ if result.Continued != false {
+ t.Error("\nUnexpected value for continued in multi batch")
+ }
+ }
+ }
+}
+
+////////////////////////////////////////
+// Test helpers to create test data
+
+// Consider a database with a list List1 with two tasks Task1, Task2. The
+// following helper methods create conflicts with these three objects in various
+// ways.
+
+// simpleConflictStream prepares a conflict stream with two conflicts, one
+// each for Task1 and Task2. No batch is involved and Task1 is independent of
+// Task2. Assume that List does not get updated for these task updates.
+// Local syncbase changes Text field for Task1 and marks Task2 done.
+// Remote syncbase marks Task1 done and Task2 done.
+// Expected result is a merge for Task1 and Task2.
+func simpleConflictStream(t *testing.T) ([]wire.ConflictInfo, map[string]wire.ResolutionInfo) {
+ // Conflict for task1
+ row1 := makeRowInfo(Task1,
+ encode(&Task{"OriginalText", false, -1, List1}), 1, // ancestor
+ encode(&Task{TextNew, false, -1, List1}), 3, // local
+ encode(&Task{"OriginalText", true, 20, List1}), 2) // remote
+ c1 := makeConflictInfo(row1, false)
+
+ // Expected result
+ r1 := makeResolution(Task1, encode(&Task{TextNew, true, 20, List1}), wire.ValueSelectionOther)
+
+ // Conflict for task2
+ row2 := makeRowInfo(Task2,
+ encode(&Task{"Text1", false, -1, List1}), 1, // ancestor
+ encode(&Task{"Text1", true, 100, List1}), 3, // local
+ encode(&Task{"Text1", true, 20, List1}), 2) // remote
+ c2 := makeConflictInfo(row2, false)
+
+ // Expected result
+ r2 := makeResolution(Task2, encode(&Task{"Text1", true, 20, List1}),
+ wire.ValueSelectionOther) // for simplicity the test resolver does not optimize
+
+ respMap := map[string]wire.ResolutionInfo{}
+ respMap[r1.Key] = r1
+ respMap[r2.Key] = r2
+
+ return []wire.ConflictInfo{c1, c2}, respMap
+}
+
+// addDeleteConflictStream prepares a conflict stream with one conflict
+// containing Task1, Task2 and List1. List contains a task count and hence
+// addition and deletion of task leads to update to the list.
+// Local syncbase deletes Task1 and updates List1 in a batch.
+// Remote syncbase adds Task2 and updates List1 in a batch.
+// Expected result is Task1 gets deleted, Task2 gets added and List1 shows
+// the correct taskcount after the two operations.
+func addDeleteConflictStream(t *testing.T) ([]wire.ConflictInfo, map[string]wire.ResolutionInfo) {
+ // Batch1 is local and contains task1 and list1
+ b1 := makeConflictBatch(1, HintTaskDelete, wire.BatchSourceLocal, true)
+
+ // Batch2 is remote and contains task2 and list1
+ b2 := makeConflictBatch(2, HintTaskAdd, wire.BatchSourceRemote, true)
+
+ // Deletion for task1 on local syncbase
+ row1 := makeRowInfo(Task1,
+ encode(&Task{"TaskToBeRemoved", false, -1, List1}), 1, // ancestor
+ nil, -1, // local
+ encode(&Task{"TaskToBeRemoved", false, -1, List1}), 1) // remote
+ row1.BatchIds = []uint16{1}
+ c1 := makeConflictInfo(row1, true)
+
+
+ // Expected result
+ r1 := makeResolution(Task1, nil, wire.ValueSelectionOther)
+
+ // Addition for task2 on remote syncbase
+ row2 := makeRowInfo(Task2,
+ nil, -1, // ancestor value
+ nil, -1, // local value
+ encode(&Task{"AddedTask", false, -1, List1}), 2) // remote value
+ row2.BatchIds = []uint16{2}
+ c2 := makeConflictInfo(row2, true)
+
+ // Expected result
+ r2 := makeResolution(Task2, encode(&Task{"AddedTask", false, -1, List1}), wire.ValueSelectionRemote)
+
+ // Update to List on both sided on filed TaskCount
+ listName := "Groceries"
+ row3 := makeRowInfo(List1,
+ encode(&List{listName, 8, 0}), 25, // ancestor value
+ encode(&List{listName, 7, 0}), 33, // local value
+ encode(&List{listName, 9, 0}), 44) // remote value
+ row3.BatchIds = []uint16{1, 2}
+ c3 := makeConflictInfo(row3, false)
+
+ // Expected result
+ r3 := makeResolution(List1, encode(&List{listName, 8, 0}), wire.ValueSelectionOther)
+
+ respMap := map[string]wire.ResolutionInfo{}
+ respMap[r1.Key] = r1
+ respMap[r2.Key] = r2
+ respMap[r3.Key] = r3
+
+ return []wire.ConflictInfo{b1, b2, c1, c2, c3}, respMap
+}
+
+// twoIntersectingBatchesConflictStream prepares a conflict stream with one
+// conflict containing Task1, Task2 and List1. Here we assume that List keeps
+// a count of done tasks.
+// Local syncbase marks Task1 done and updates done count on List1. Later it
+// it also marks Task2 done and updates done count on List1.
+// Remote syncbase updates field Text on Task1 followed by an update to field
+// Text on Task2 (not in a batch).
+// Expected result is a merge for Task1 and Task2 and Local version of List1
+// being accepted.
+func twoIntersectingBatchesConflictStream(t *testing.T) ([]wire.ConflictInfo, map[string]wire.ResolutionInfo) {
+ // Batch1 is local and contains task1 and list1
+ b1 := makeConflictBatch(1, HintDone, wire.BatchSourceLocal, true)
+
+ // Batch2 is local and contains task2 and list1
+ b2 := makeConflictBatch(2, HintDone, wire.BatchSourceLocal, true)
+
+ // Batch3 is remote and contains task1
+ b3 := makeConflictBatch(3, HintTextEdit, wire.BatchSourceRemote, true)
+
+ // Batch4 is remote and contains task2
+ b4 := makeConflictBatch(4, HintTextEdit, wire.BatchSourceRemote, true)
+
+ // For task1, mark done on local and edit text on remote
+ row1 := makeRowInfo(Task1,
+ encode(&Task{"TaskOrig", false, -1, List1}), 1, // ancestor
+ encode(&Task{"TaskOrig", true, 204, List1}), 5, // local
+ encode(&Task{"TaskEdit", false, -1, List1}), 3) // remote
+ row1.BatchIds = []uint16{1, 3}
+ c1 := makeConflictInfo(row1, true)
+
+ // Expected result
+ r1 := makeResolution(Task1, encode(&Task{"TaskEdit", true, 204, List1}), wire.ValueSelectionOther)
+
+ // For task2, mark done on local and edit text on remote
+ row2 := makeRowInfo(Task2,
+ encode(&Task{"TaskOrig", false, -1, List1}), 2, // ancestor
+ encode(&Task{"TaskOrig", true, 204, List1}), 5, // local
+ encode(&Task{"TaskEdit", false, -1, List1}), 3) // remote
+ row2.BatchIds = []uint16{2, 4}
+ c2 := makeConflictInfo(row2, true)
+
+ // Expected result
+ r2 := makeResolution(Task2, encode(&Task{"TaskEdit", true, 204, List1}), wire.ValueSelectionOther)
+
+ // Update to List on Local for updating TaskDoneCount
+ listName := "Groceries"
+ row3 := makeRowInfo(List1,
+ encode(&List{listName, 8, 0}), 1, // ancestor value
+ encode(&List{listName, 8, 2}), 5, // local value
+ encode(&List{listName, 8, 0}), 3) // remote value
+ row3.BatchIds = []uint16{1, 2}
+ c3 := makeConflictInfo(row3, false)
+
+ // Expected result
+ r3 := makeResolution(List1, encode(&List{listName, 8, 2}), wire.ValueSelectionOther)
+
+ respMap := map[string]wire.ResolutionInfo{}
+ respMap[r1.Key] = r1
+ respMap[r2.Key] = r2
+ respMap[r3.Key] = r3
+
+ return []wire.ConflictInfo{b1, b2, b3, b4, c1, c2, c3}, respMap
+}
+
+// listDeleteConflictsWithTaskAdd prepares a conflict stream with one
+// conflict containing Task1, Task2 and List1. Assume Task2 does not exist yet.
+// Local syncbase deletes List1 and Task1 as a batch.
+// Remote syncbase adds Task2 along with an update to List1 as a batch.
+// Expected result is deletion of List1, Task1 and Task2.
+func listDeleteConflictsWithTaskAdd(t *testing.T) ([]wire.ConflictInfo, map[string]wire.ResolutionInfo) {
+ // Batch1 is local and contains task1 and list1
+ b1 := makeConflictBatch(1, HintListDelete, wire.BatchSourceLocal, true)
+
+ // Batch2 is remote and contains task2 and list1
+ b2 := makeConflictBatch(2, HintTaskAdd, wire.BatchSourceRemote, true)
+
+ // Delition for list1 on local syncbase
+ listName := "Groceries"
+ row0 := makeRowInfo(List1,
+ encode(&List{listName, 1, 0}), 1, // ancestor value
+ nil, 5, // local value
+ encode(&List{listName, 2, 0}), 3) // remote value
+ row0.BatchIds = []uint16{1, 2}
+ c0 := makeConflictInfo(row0, true)
+
+ // Expected result
+ r0 := makeResolution(List1, nil, wire.ValueSelectionOther)
+
+ // Deletion for task1 on local syncbase along with list1
+ row1 := makeRowInfo(Task1,
+ encode(&Task{"TaskToBeRemoved", false, -1, List1}), 1, // ancestor
+ nil, -1, // local
+ encode(&Task{"TaskToBeRemoved", false, -1, List1}), 1) // remote
+ row1.BatchIds = []uint16{1}
+ c1 := makeConflictInfo(row1, true)
+
+ // Expected result
+ r1 := makeResolution(Task1, nil, wire.ValueSelectionOther)
+
+ // Addition for task2 on remote syncbase
+ row2 := makeRowInfo(Task2,
+ nil, -1, // ancestor value
+ nil, -1, // local value
+ encode(&Task{"AddedTask", false, -1, List1}), 2) // remote value
+ row2.BatchIds = []uint16{2}
+ c2 := makeConflictInfo(row2, false)
+
+ // Expected result
+ r2 := makeResolution(Task2, nil, wire.ValueSelectionOther)
+
+ respMap := map[string]wire.ResolutionInfo{}
+ respMap[r0.Key] = r0
+ respMap[r1.Key] = r1
+ respMap[r2.Key] = r2
+
+ return []wire.ConflictInfo{b1, b2, c0, c1, c2}, respMap
+}
+
+func makeResolution(key string, result []byte, selection wire.ValueSelection) wire.ResolutionInfo {
+ r := wire.ResolutionInfo{}
+ r.Key = key
+ if result != nil {
+ r.Result = &wire.Value{
+ Bytes: result,
+ WriteTs: -1,
+ }
+ }
+ r.Selection = selection
+ return r
+}
+
+func makeConflictBatch(id uint16, hint string, source wire.BatchSource, continued bool) wire.ConflictInfo {
+ batch := wire.BatchInfo{Id: id, Hint: hint, Source: source}
+ return wire.ConflictInfo{Data: wire.ConflictDataBatch{Value: batch}, Continued: continued}
+}
+
+func makeConflictInfo(row wire.RowInfo, continued bool) wire.ConflictInfo {
+ return wire.ConflictInfo{
+ Data: wire.ConflictDataRow{Value: row},
+ Continued: continued,
+ }
+}
+
+func makeRowInfo(key string, ancestor []byte, ats int64, local []byte, lts int64, remote []byte, rts int64) wire.RowInfo {
+ op := wire.RowOp{}
+ op.Key = key
+
+ if ancestor != nil {
+ op.AncestorValue = &wire.Value{
+ Bytes: ancestor,
+ WriteTs: ats,
+ }
+ }
+
+ if local != nil {
+ op.LocalValue = &wire.Value{
+ Bytes: local,
+ WriteTs: lts,
+ }
+ }
+
+ if remote != nil {
+ op.RemoteValue = &wire.Value{
+ Bytes: remote,
+ WriteTs: rts,
+ }
+ }
+ return wire.RowInfo{
+ Op: wire.OperationWrite{Value: op},
+ }
+}
+
+func compareResult(t *testing.T, expected wire.ResolutionInfo, actual wire.ResolutionInfo) {
+ if actual.Key != expected.Key {
+ t.Error("Key does not match")
+ }
+ if actual.Selection != expected.Selection {
+ t.Errorf("Key: %s", expected.Key)
+ t.Errorf("Expected selection: %v, actual selection: %v", expected.Selection, actual.Selection)
+ }
+ if (expected.Result == nil) && (actual.Result != nil) {
+ t.Errorf("Key: %s", expected.Key)
+ t.Error("Result expected to be nil but found non nil")
+ }
+ if expected.Result != nil {
+ if actual.Result == nil {
+ t.Errorf("Key: %s", expected.Key)
+ t.Error("Result found nil")
+ }
+ if bytes.Compare(actual.Result.Bytes, expected.Result.Bytes) != 0 {
+ t.Errorf("Key: %s", expected.Key)
+ t.Error("Result bytes do not match")
+ }
+ }
+}
+
+func encode(value interface{}) []byte {
+ v, _ := vom.Encode(value)
+ return v
+}
diff --git a/v23/syncbase/nosql/crtestutil/mock_dbclient.go b/v23/syncbase/nosql/crtestutil/mock_dbclient.go
new file mode 100644
index 0000000..99e27ee
--- /dev/null
+++ b/v23/syncbase/nosql/crtestutil/mock_dbclient.go
@@ -0,0 +1,34 @@
+// 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 crtestutil
+
+import (
+ wire "v.io/syncbase/v23/services/syncbase/nosql"
+ "v.io/v23/context"
+ "v.io/v23/rpc"
+ "v.io/v23/verror"
+)
+
+type MockWireDatabaseClient struct {
+ wire.DatabaseClientMethods
+ crs *CrStreamImpl
+}
+
+func MockDbClient(c wire.DatabaseClientMethods, crs *CrStreamImpl) *MockWireDatabaseClient {
+ return &MockWireDatabaseClient{c, crs}
+}
+
+func (wclient *MockWireDatabaseClient) GetSchemaMetadata(ctx *context.T, opts ...rpc.CallOpt) (sm wire.SchemaMetadata, err error) {
+ err = verror.NewErrNoExist(ctx)
+ return
+}
+
+func (wclient *MockWireDatabaseClient) SetSchemaMetadata(ctx *context.T, i0 wire.SchemaMetadata, opts ...rpc.CallOpt) error {
+ return nil
+}
+
+func (wclient *MockWireDatabaseClient) StartConflictResolver(ctx *context.T, opts ...rpc.CallOpt) (wire.ConflictManagerStartConflictResolverClientCall, error) {
+ return wclient.crs, nil
+}
diff --git a/v23/syncbase/nosql/crtestutil/mock_stream.go b/v23/syncbase/nosql/crtestutil/mock_stream.go
new file mode 100644
index 0000000..3bb8f4d
--- /dev/null
+++ b/v23/syncbase/nosql/crtestutil/mock_stream.go
@@ -0,0 +1,119 @@
+// 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 crtestutil
+
+import (
+ "sync"
+
+ wire "v.io/syncbase/v23/services/syncbase/nosql"
+)
+
+var _ wire.ConflictManagerStartConflictResolverClientCall = (*CrStreamImpl)(nil)
+
+type State struct {
+ // ConflictStream state
+ IsBlocked bool
+ Mu sync.Mutex
+ Val wire.ConflictInfo
+ AdvanceCount int
+ ValIndex int
+ // ResolutionStream state variables
+ Result []wire.ResolutionInfo
+}
+
+type CrStreamImpl struct {
+ C ConflictStream
+ R ResolutionStream
+}
+
+func (s *CrStreamImpl) RecvStream() interface {
+ Advance() bool
+ Value() wire.ConflictInfo
+ Err() error
+} {
+ return recvStreamImpl{s.C}
+}
+
+func (s *CrStreamImpl) SendStream() interface {
+ Send(item wire.ResolutionInfo) error
+ Close() error
+} {
+ return sendStreamImpl{s.R}
+}
+
+func (s *CrStreamImpl) Finish() error {
+ return nil
+}
+
+type recvStreamImpl struct {
+ c ConflictStream
+}
+
+func (rs recvStreamImpl) Advance() bool {
+ return rs.c.Advance()
+}
+func (rs recvStreamImpl) Value() wire.ConflictInfo {
+ return rs.c.Value()
+}
+func (rs recvStreamImpl) Err() error {
+ return rs.c.Err()
+}
+
+type sendStreamImpl struct {
+ r ResolutionStream
+}
+
+func (ss sendStreamImpl) Send(item wire.ResolutionInfo) error {
+ return ss.r.Send(item)
+}
+func (c sendStreamImpl) Close() error {
+ return nil
+}
+
+type ConflictStream interface {
+ Advance() bool
+ Value() wire.ConflictInfo
+ Err() error
+}
+
+type ResolutionStream interface {
+ Send(item wire.ResolutionInfo) error
+}
+
+type ConflictStreamImpl struct {
+ St *State
+ AdvanceFn func(*State) bool
+}
+
+func (cs *ConflictStreamImpl) Advance() bool {
+ return cs.AdvanceFn(cs.St)
+}
+func (cs *ConflictStreamImpl) Value() wire.ConflictInfo {
+ return cs.St.Val
+}
+func (cs *ConflictStreamImpl) Err() error {
+ return &TestError{"Stream broken"}
+}
+
+type ResolutionStreamImpl struct {
+ St *State
+}
+
+func (rs *ResolutionStreamImpl) Send(item wire.ResolutionInfo) error {
+ if rs.St.Result == nil {
+ rs.St.Result = []wire.ResolutionInfo{item}
+ return nil
+ }
+ rs.St.Result = append(rs.St.Result, item)
+ return nil
+}
+
+type TestError struct {
+ str string
+}
+
+func (e *TestError) Error() string {
+ return e.str
+}
diff --git a/v23/syncbase/nosql/database.go b/v23/syncbase/nosql/database.go
index d48f450..42b5270 100644
--- a/v23/syncbase/nosql/database.go
+++ b/v23/syncbase/nosql/database.go
@@ -5,6 +5,9 @@
package nosql
import (
+ "sync"
+ "time"
+
wire "v.io/syncbase/v23/services/syncbase/nosql"
"v.io/syncbase/v23/syncbase/util"
"v.io/v23/context"
@@ -15,6 +18,12 @@
"v.io/x/lib/vlog"
)
+const (
+ // Wait time before we try to reconnect a broken conflict resolution stream.
+ waitBeforeReconnectInMillis = 2 * time.Second
+ reconnectionCount = "rcc"
+)
+
func NewDatabase(parentFullName, relativeName string, schema *Schema) *database {
fullName := naming.Join(parentFullName, relativeName)
return &database{
@@ -23,6 +32,9 @@
fullName: fullName,
name: relativeName,
schema: schema,
+ crState: conflictResolutionState{
+ reconnectWaitTime: waitBeforeReconnectInMillis,
+ },
}
}
@@ -32,6 +44,31 @@
fullName string
name string
schema *Schema
+ crState conflictResolutionState
+}
+
+// conflictResolutionState maintains data about the connection of
+// conflict resolution stream with syncbase. It provides a way to disconnect
+// an existing open stream.
+type conflictResolutionState struct {
+ mu sync.Mutex // guards access to all fields in this struct
+ crContext *context.T
+ cancelFn context.CancelFunc
+ isClosed bool
+ reconnectWaitTime time.Duration
+}
+
+func (crs *conflictResolutionState) disconnect() {
+ crs.mu.Lock()
+ defer crs.mu.Unlock()
+ crs.isClosed = true
+ crs.cancelFn()
+}
+
+func (crs *conflictResolutionState) isDisconnected() bool {
+ crs.mu.Lock()
+ defer crs.mu.Unlock()
+ return crs.isClosed
}
var _ Database = (*database)(nil)
@@ -168,17 +205,44 @@
return createBlob(ctx, d.fullName)
}
-// UpgradeIfOutdated implements Database.UpgradeIfOutdated.
-func (d *database) UpgradeIfOutdated(ctx *context.T) (bool, error) {
+// EnforceSchema implements Database.EnforceSchema.
+func (d *database) EnforceSchema(ctx *context.T) error {
var schema *Schema = d.schema
if schema == nil {
- return false, verror.New(verror.ErrBadState, ctx, "Schema or SchemaMetadata cannot be nil. A valid Schema needs to be used when creating DB handle.")
+ return verror.New(verror.ErrBadState, ctx, "Schema or SchemaMetadata cannot be nil. A valid Schema needs to be used when creating DB handle.")
}
if schema.Metadata.Version < 0 {
- return false, verror.New(verror.ErrBadState, ctx, "Schema version cannot be less than zero.")
+ return verror.New(verror.ErrBadState, ctx, "Schema version cannot be less than zero.")
}
+ if needsResolver(d.schema.Metadata) && d.schema.Resolver == nil {
+ return verror.New(verror.ErrBadState, ctx, "ResolverTypeAppResolves cannot be used in CrRule without providing a ConflictResolver in Schema.")
+ }
+
+ if _, err := d.upgradeIfOutdated(ctx); err != nil {
+ return err
+ }
+
+ if d.schema.Resolver == nil {
+ return nil
+ }
+
+ childCtx, cancelFn := context.WithCancel(ctx)
+ d.crState.crContext = childCtx
+ d.crState.cancelFn = cancelFn
+
+ go d.establishConflictResolution(childCtx)
+ return nil
+}
+
+// Close implements Database.Close.
+func (d *database) Close() {
+ d.crState.disconnect()
+}
+
+func (d *database) upgradeIfOutdated(ctx *context.T) (bool, error) {
+ var schema *Schema = d.schema
schemaMgr := d.getSchemaManager()
currMeta, err := schemaMgr.getSchemaMetadata(ctx)
if err != nil {
@@ -222,8 +286,182 @@
return true, nil
}
+func (d *database) establishConflictResolution(ctx *context.T) {
+ count := 0
+ for {
+ count++
+ vlog.Infof("Starting a new conflict resolution connection. Re-Connection count: %d", count)
+ childCtx := context.WithValue(ctx, reconnectionCount, count)
+ // listenForConflicts is a blocking method which returns only when the
+ // conflict stream is broken.
+ if err := d.listenForConflicts(childCtx); err != nil {
+ vlog.Errorf("Conflict resolution connection ended with error: %v", err)
+ }
+
+ // Check if database is closed and if we need to shutdown conflict
+ // resolution.
+ if d.crState.isDisconnected() {
+ vlog.Infof("Shutting down conflict resolution connection.")
+ break
+ }
+
+ // The connection might have broken because syncbase service went down.
+ // Sleep for a few seconds to allow syncbase to come back up.
+ time.Sleep(d.crState.reconnectWaitTime)
+ }
+}
+
+func (d *database) listenForConflicts(ctx *context.T) error {
+ resolver, err := d.c.StartConflictResolver(ctx)
+ if err != nil {
+ return err
+ }
+ conflictStream := resolver.RecvStream()
+ resolutionStream := resolver.SendStream()
+ var c *Conflict = &Conflict{}
+ for conflictStream.Advance() {
+ row := conflictStream.Value()
+ addRowToConflict(c, &row)
+ if !row.Continued {
+ resolution := d.schema.Resolver.OnConflict(ctx, c)
+ if err := sendResolution(resolutionStream, resolution); err != nil {
+ return err
+ }
+ c = &Conflict{} // create a new conflict object for the next batch
+ }
+ }
+ if err := conflictStream.Err(); err != nil {
+ return err
+ }
+ return resolver.Finish()
+}
+
+// TODO(jlodhia): Should we check if the Resolution received addresses all
+// conflicts in write set?
+func sendResolution(stream interface {
+ Send(item wire.ResolutionInfo) error
+}, resolution Resolution) error {
+ size := len(resolution.ResultSet)
+ count := 0
+ for _, v := range resolution.ResultSet {
+ count++
+ ri := toResolutionInfo(v, count != size)
+ if err := stream.Send(ri); err != nil {
+ vlog.Error("Error while sending resolution")
+ return err
+ }
+ }
+ return nil
+}
+
+func addRowToConflict(c *Conflict, ci *wire.ConflictInfo) {
+ switch v := ci.Data.(type) {
+ case wire.ConflictDataBatch:
+ if c.Batches == nil {
+ c.Batches = map[uint16]wire.BatchInfo{}
+ }
+ c.Batches[v.Value.Id] = v.Value
+ case wire.ConflictDataRow:
+ rowInfo := v.Value
+ switch op := rowInfo.Op.(type) {
+ case wire.OperationWrite:
+ if c.WriteSet == nil {
+ c.WriteSet = &ConflictRowSet{map[string]ConflictRow{}, map[uint16][]ConflictRow{}}
+ }
+ cr := toConflictRow(op.Value, rowInfo.BatchIds)
+ c.WriteSet.ByKey[cr.Key] = cr
+ for _, bid := range rowInfo.BatchIds {
+ c.WriteSet.ByBatch[bid] = append(c.WriteSet.ByBatch[bid], cr)
+ }
+ case wire.OperationRead:
+ if c.ReadSet == nil {
+ c.ReadSet = &ConflictRowSet{map[string]ConflictRow{}, map[uint16][]ConflictRow{}}
+ }
+ cr := toConflictRow(op.Value, rowInfo.BatchIds)
+ c.ReadSet.ByKey[cr.Key] = cr
+ for _, bid := range rowInfo.BatchIds {
+ c.ReadSet.ByBatch[bid] = append(c.ReadSet.ByBatch[bid], cr)
+ }
+ case wire.OperationScan:
+ if c.ScanSet == nil {
+ c.ScanSet = &ConflictScanSet{map[uint16][]wire.ScanOp{}}
+ }
+ for _, bid := range rowInfo.BatchIds {
+ c.ScanSet.ByBatch[bid] = append(c.ScanSet.ByBatch[bid], op.Value)
+ }
+ }
+ }
+}
+
+func toConflictRow(op wire.RowOp, batchIds []uint16) ConflictRow {
+ var local, remote, ancestor *Value
+ if op.LocalValue != nil {
+ local = &Value{
+ val: op.LocalValue.Bytes,
+ WriteTs: toTime(op.LocalValue.WriteTs),
+ selection: wire.ValueSelectionLocal,
+ }
+ }
+ if op.RemoteValue != nil {
+ remote = &Value{
+ val: op.RemoteValue.Bytes,
+ WriteTs: toTime(op.RemoteValue.WriteTs),
+ selection: wire.ValueSelectionRemote,
+ }
+ }
+ if op.AncestorValue != nil {
+ ancestor = &Value{
+ val: op.AncestorValue.Bytes,
+ WriteTs: toTime(op.AncestorValue.WriteTs),
+ selection: wire.ValueSelectionOther,
+ }
+ }
+ return ConflictRow{
+ Key: op.Key,
+ LocalValue: local,
+ RemoteValue: remote,
+ AncestorValue: ancestor,
+ BatchIds: batchIds,
+ }
+}
+
+// TODO(jlodhia): remove this method once time is stored as time.Time instead
+// of int64
+func toTime(unixNanos int64) time.Time {
+ return time.Unix(
+ unixNanos / 1e9, // seconds
+ unixNanos % 1e9) // nanoseconds
+}
+
+func toResolutionInfo(r ResolvedRow, lastRow bool) wire.ResolutionInfo {
+ sel := wire.ValueSelectionOther
+ resVal := (*wire.Value)(nil)
+ if r.Result != nil {
+ sel = r.Result.selection
+ resVal = &wire.Value{
+ Bytes: r.Result.val,
+ WriteTs: r.Result.WriteTs.UnixNano(), // this timestamp is ignored by syncbase
+ }
+ }
+ return wire.ResolutionInfo{
+ Key: r.Key,
+ Selection: sel,
+ Result: resVal,
+ Continued: lastRow,
+ }
+}
+
+func needsResolver(metadata wire.SchemaMetadata) bool {
+ for _, rule := range metadata.Policy.Rules {
+ if rule.Resolver == wire.ResolverTypeAppResolves {
+ return true
+ }
+ }
+ return false
+}
+
func (d *database) getSchemaManager() schemaManagerImpl {
- return newSchemaManager(d.fullName)
+ return newSchemaManager(d.c)
}
func (d *database) schemaVersion() int32 {
diff --git a/v23/syncbase/nosql/model.go b/v23/syncbase/nosql/model.go
index 622f82d..d9ae93f 100644
--- a/v23/syncbase/nosql/model.go
+++ b/v23/syncbase/nosql/model.go
@@ -6,12 +6,15 @@
package nosql
import (
+ "time"
+
wire "v.io/syncbase/v23/services/syncbase/nosql"
"v.io/syncbase/v23/syncbase/util"
"v.io/v23/context"
"v.io/v23/security/access"
"v.io/v23/services/watch"
"v.io/v23/vdl"
+ "v.io/v23/verror"
"v.io/v23/vom"
)
@@ -45,6 +48,10 @@
// time of the RPC, and will not reflect subsequent writes to keys not yet
// reached by the stream.
Exec(ctx *context.T, query string) ([]string, ResultStream, error)
+
+ // Close cleans up any state associated with this client handle including
+ // shutting down any open conflict resolution stream.
+ Close()
}
// Database represents a collection of Tables. Batches, queries, sync, watch,
@@ -131,14 +138,15 @@
// Blob returns a handle to the blob with the given BlobRef.
Blob(br wire.BlobRef) Blob
- // This method compares the current schema version of the database with the
- // schema version provided while creating this database handle. If the
- // current database schema version is lower, then the SchemaUpdater is
+ // EnforceSchema compares the current schema version of the database
+ // with the schema version provided while creating this database handle. If
+ // the current database schema version is lower, then the SchemaUpdater is
// called. If SchemaUpdater is successful this method stores the new schema
// metadata in database.
- // Note: schema can be nil, in which case this method skips schema check
- // and the caller is responsible for maintaining schema sanity.
- UpgradeIfOutdated(ctx *context.T) (bool, error)
+ // This method also registers a conflict resolver with syncbase to receive
+ // conflicts. Note: schema can be nil, in which case this method skips
+ // schema check and the caller is responsible for maintaining schema sanity.
+ EnforceSchema(ctx *context.T) error
}
// BatchDatabase is a handle to a set of reads and writes to the database that
@@ -550,4 +558,114 @@
type Schema struct {
Metadata wire.SchemaMetadata
Upgrader SchemaUpgrader
+ Resolver ConflictResolver
+}
+
+// ConflictResolver interface allows the App to define resolution of conflicts
+// that it requested to handle.
+type ConflictResolver interface {
+ OnConflict(ctx *context.T, conflict *Conflict) Resolution
+}
+
+// Conflict contains information to fully specify a conflict. Since syncbase
+// supports batches there can be one or more rows within the batch that has a
+// conflict. Each of these rows will be sent together as part of a single
+// conflict. Each row contains an Id of the batch to which it belongs,
+// enabling the client to group together rows that are part of a batch. Note
+// that a single row can be part of more than one batch.
+//
+// WriteSet contains rows that were written.
+// ReadSet contains rows that were read within a batch corresponding to a row
+// within the write set.
+// ScanSet contains scans performed within a batch corresponding to a row
+// within the write set.
+// Batches is a map of unique ids to BatchInfo objects. The id is unique only in
+// the context of a given conflict and is otherwise meaningless.
+type Conflict struct {
+ ReadSet *ConflictRowSet
+ WriteSet *ConflictRowSet
+ ScanSet *ConflictScanSet
+ Batches map[uint16]wire.BatchInfo
+}
+
+// ConflictRowSet contains a set of rows under conflict. It provides two different
+// ways to access the same set.
+// ByKey is a map of ConflictRows keyed by the row key.
+// ByBatch is a map of []ConflictRows keyed by batch id. This map lets the client
+// access all ConflictRows within this set that contain a given hint.
+type ConflictRowSet struct {
+ ByKey map[string]ConflictRow
+ ByBatch map[uint16][]ConflictRow
+}
+
+// ConflictScanSet contains a set of scans under conflict.
+// ByBatch is a map of array of ScanOps keyed by batch id.
+type ConflictScanSet struct {
+ ByBatch map[uint16][]wire.ScanOp
+}
+
+// ConflictRow represents a row under conflict.
+// Key is the key for the row.
+// LocalValue is the value present in the local db.
+// RemoteValue is the value received via sync.
+// AncestorValue is the value for the key which is the lowest common
+// ancestor of the two values represented by LocalValue and RemoteValue.
+// AncestorValue is nil if the ConflictRow is a part of the read set.
+// BatchIds is a list of ids of all the batches that this row belongs to.
+type ConflictRow struct {
+ Key string
+ LocalValue *Value
+ RemoteValue *Value
+ AncestorValue *Value
+ BatchIds []uint16
+}
+
+// Resolution contains the application’s reply to a conflict. It must contain a
+// resolved value for each conflict row within the WriteSet of the given
+// conflict.
+// ResultSet is a map of row key to ResolvedRow.
+type Resolution struct {
+ ResultSet map[string]ResolvedRow
+ // TODO(jlodhia): Hint []string
+}
+
+// ResolvedRow represents a result of resolution of a row under conflict.
+// Key is the key for the row.
+// Selection represents the value that was selected for resolution.
+// Value is the resolved value for the key. This field should be used only
+// if value of Selection field is 'Other'.
+type ResolvedRow struct {
+ Key string
+ Result *Value
+}
+
+// Value contains a specific version of data for the row under conflict along
+// with the write timestamp and hints associated with the version.
+// WriteTs is the write timestamp for this value.
+type Value struct {
+ val []byte
+ WriteTs time.Time
+ selection wire.ValueSelection
+}
+
+// Get takes a reference to an instance of a type that is expected to be
+// represented by Value.
+func (v *Value) Get(value interface{}) error {
+ return vom.Decode(v.val, value)
+}
+
+// NewValue creates a new Value to be added to Resolution.
+func NewValue(ctx *context.T, data interface{}) (*Value, error) {
+ if data == nil {
+ return nil, verror.New(verror.ErrBadArg, ctx, "data cannot be nil")
+ }
+ bytes, err := vom.Encode(data)
+ if err != nil {
+ return nil, err
+ }
+ return &Value{
+ val: bytes,
+ WriteTs: time.Now(), // ignored by syncbase
+ selection: wire.ValueSelectionOther,
+ }, nil
}
diff --git a/v23/syncbase/nosql/schema.go b/v23/syncbase/nosql/schema.go
index 9f52a1d..8dd480d 100644
--- a/v23/syncbase/nosql/schema.go
+++ b/v23/syncbase/nosql/schema.go
@@ -21,14 +21,12 @@
// Implementation of SchemaManager (Not part of public client API)
type schemaManagerImpl struct {
- dbName string
- c wire.DatabaseClientMethods
+ c wire.DatabaseClientMethods
}
-func newSchemaManager(dbName string) schemaManagerImpl {
+func newSchemaManager(client wire.DatabaseClientMethods) schemaManagerImpl {
return schemaManagerImpl{
- dbName: dbName,
- c: wire.DatabaseClient(dbName),
+ c: client,
}
}
diff --git a/v23/syncbase/nosql/schema_test.go b/v23/syncbase/nosql/schema_test.go
index 6d7ed0d..78ebd95 100644
--- a/v23/syncbase/nosql/schema_test.go
+++ b/v23/syncbase/nosql/schema_test.go
@@ -19,10 +19,10 @@
// This test as following steps:
// 1) Call NoSQLDatabase() for a non existent db.
// 2) Create the database, and verify if Schema got stored properly.
-// 3) Call UpgradeIfOutdated() to make sure that the method is no-op and is
+// 3) Call EnforceSchema() to make sure that the method is no-op and is
// able to read the schema from db.
// 4) Call NoSQLDatabase() on the same db to create a new handle with an
-// upgraded schema, call UpgradeIfOutdated() and check if SchemaUpgrader
+// upgraded schema, call EnforceSchema() and check if SchemaUpgrader
// is called and if the new schema is stored appropriately.
func TestSchemaCheck(t *testing.T) {
ctx, sName, cleanup := tu.SetupOrDie(nil)
@@ -35,9 +35,9 @@
// Verify that calling Upgrade on a non existing database does not throw
// errors.
- _, err := db1.UpgradeIfOutdated(ctx)
+ err := db1.EnforceSchema(ctx)
if err != nil {
- t.Fatalf("db1.UpgradeIfOutdated() failed: %v", err)
+ t.Fatalf("db1.EnforceSchema() failed: %v", err)
}
if mockUpgrader.CallCount > 0 {
t.Fatal("Call to upgrader was not expected.")
@@ -53,12 +53,8 @@
}
// Make redundant call to Upgrade to verify that it is a no-op
- result, err1 := db1.UpgradeIfOutdated(ctx)
- if result {
- t.Fatalf("db1.UpgradeIfOutdated() should not return true")
- }
- if err1 != nil {
- t.Fatalf("db1.UpgradeIfOutdated() failed: %v", err1)
+ if err := db1.EnforceSchema(ctx); err != nil {
+ t.Fatalf("db1.EnforceSchema() failed: %v", err)
}
if mockUpgrader.CallCount > 0 {
t.Fatal("Call to upgrader was not expected.")
@@ -73,13 +69,8 @@
}
schema.Metadata.Policy = policy
otherdb1 := a.NoSQLDatabase("db1", schema)
- otherresult, othererr := otherdb1.UpgradeIfOutdated(ctx)
-
- if !otherresult {
- t.Fatalf("otherdb1.UpgradeIfOutdated() expected to return true")
- }
- if othererr != nil {
- t.Fatalf("otherdb1.UpgradeIfOutdated() failed: %v", othererr)
+ if err := otherdb1.EnforceSchema(ctx); err != nil {
+ t.Fatalf("otherdb1.EnforceSchema() failed: %v", err)
}
if mockUpgrader.CallCount != 1 {
t.Fatalf("Unexpected number of calls to upgrader. Expected: %d, Actual: %d.", 1, mockUpgrader.CallCount)
@@ -226,7 +217,7 @@
// Upgrade schema version for underlying db using a different handle
schema2 := tu.DefaultSchema(1)
dbHandle2 := a.NoSQLDatabase("db1", schema2)
- dbHandle2.UpgradeIfOutdated(ctx)
+ dbHandle2.EnforceSchema(ctx)
// Commit batch1, abort batch2, attempt writing a row using batch3.
// Each of these operations should fail.
diff --git a/v23/syncbase/util/util.go b/v23/syncbase/util/util.go
index 6eeaaaa..dda4c6d 100644
--- a/v23/syncbase/util/util.go
+++ b/v23/syncbase/util/util.go
@@ -64,6 +64,12 @@
return string(x)
}
+// IsPrefix returns true if start and limit strings together represent a prefix
+// range. If true, start represents the prefix.
+func IsPrefix(start string, limit string) bool {
+ return PrefixRangeLimit(start) == limit
+}
+
// AccessController provides access control for various syncbase objects.
type AccessController interface {
// SetPermissions replaces the current Permissions for an object.
diff --git a/v23/syncbase/util/util_test.go b/v23/syncbase/util/util_test.go
index 321691a..2d6384a 100644
--- a/v23/syncbase/util/util_test.go
+++ b/v23/syncbase/util/util_test.go
@@ -61,3 +61,42 @@
}
}
}
+
+func TestIsPrefix(t *testing.T) {
+ tests := []struct {
+ isPrefix bool
+ start string
+ limit string
+ }{
+ {true, "", ""},
+ {true, "a", "b"},
+ {true, "aa", "ab"},
+ {true, "\xfe", "\xff"},
+ {true, "a\xfe", "a\xff"},
+ {true, "aa\xfe", "aa\xff"},
+ {true, "a\xff", "b"},
+ {true, "aa\xff", "ab"},
+ {true, "a\xff\xff", "b"},
+ {true, "aa\xff\xff", "ab"},
+ {true, "\xff", ""},
+ {true, "\xff\xff", ""},
+
+ {false, "", "\x00"},
+ {false, "a", "aa"},
+ {false, "aa", "aa"},
+ {false, "\xfe", "\x00"},
+ {false, "a\xfe", "b\xfe"},
+ {false, "aa\xfe", "aa\x00"},
+ {false, "a\xff", "b\x00"},
+ {false, "aa\xff", "ab\x00"},
+ {false, "a\xff\xff", "a\xff\xff\xff"},
+ {false, "aa\xff\xff", "a"},
+ {false, "\xff", "\x00"},
+ }
+ for _, test := range tests {
+ result := util.IsPrefix(test.start, test.limit)
+ if result != test.isPrefix {
+ t.Errorf("%q, %q: got %v, want %v", test.start, test.limit, result, test.isPrefix)
+ }
+ }
+}
diff --git a/x/ref/services/syncbase/server/nosql/database.go b/x/ref/services/syncbase/server/nosql/database.go
index 2818a8c..a9f503f 100644
--- a/x/ref/services/syncbase/server/nosql/database.go
+++ b/x/ref/services/syncbase/server/nosql/database.go
@@ -49,6 +49,11 @@
mu sync.Mutex // protects the fields below
sns map[uint64]store.Snapshot
txs map[uint64]store.Transaction
+
+ // Active ConflictResolver connection from the app to this database.
+ // NOTE: For now, we assume there's only one open conflict resolution stream
+ // per database (typically, from the app that owns the database).
+ resolver wire.ConflictManagerStartConflictResolverServerCall
}
// databaseReq is a per-request object that handles Database RPCs.
diff --git a/x/ref/services/syncbase/server/nosql/database_crm.go b/x/ref/services/syncbase/server/nosql/database_crm.go
new file mode 100644
index 0000000..e1135a9
--- /dev/null
+++ b/x/ref/services/syncbase/server/nosql/database_crm.go
@@ -0,0 +1,20 @@
+// 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 nosql
+
+import (
+ wire "v.io/syncbase/v23/services/syncbase/nosql"
+ "v.io/v23/context"
+)
+
+////////////////////////////////////////
+// ConflictManager RPC methods
+
+func (d *databaseReq) StartConflictResolver(ctx *context.T, call wire.ConflictManagerStartConflictResolverServerCall) error {
+ // Store the conflict resolver connection in the per-app, per-database
+ // singleton so that sync can access it.
+ d.database.resolver = call
+ return nil
+}
