ING-1072: Added initial support for XDCR RPCs.

Author: Brett Lawson

docs/XDCR.md

@@ -0,0 +1,56 @@
+# XDCR Protostellar Behaviour
+
+### PushDocument
+
+PushDocument exposes two values for CAS, one is the CheckCas and one is the StoreCAS. The
+CheckCas represents the typical Compare And Swap operation of other memcached commands. If
+a CheckCas is specified, conflict resolution is disabled and assuming the CAS check completes,
+the write will be accepted. If CheckCas is set to the value of '0' explicitly, it is expected
+that the document must not exist and a AddWithMeta will be performed. StoreCAS represents the
+new CAS value for the document being pushed, and in cases where conflict resolution is in use,
+this is the value that is compared for conflict resolution.
+
+A pseudocode explanation of the logic is:
+
+```
+if req.CheckCas != nil:
+  if *req.CheckCas == 0:
+    AddWithMeta()
+  else:
+    SetWithMeta(CheckCas: req.CheckCas, Options: SkipConflictResolution)
+else:
+  SetWithMeta(CheckCas: 0, Options: 0)
+```
+
+Additionally, if IsDeleted is specified on the PushDocument operation, it is assumed that this
+is an attempt to delete the document, and as such CheckCas will optionally control whether a
+pure CAS check is performed (and 0 is an invalid value).
+
+There are a number of conflict related errors that can be returned by the PushDocument operation
+based on a number of different factors. Namely whether or not CheckCas is being applied as well as
+whether conflict logging is enabled for the operation.
+
+ConflictLogging=true:
+
+- Error:Aborted Code:TRUE_CONFLICT
+  This case will occur when the operation would otherwise succeed, but a true conflict was detected
+  and the client has not provided an explicit CAS value (to verify they have the existing document
+  already available to them locally). This error will also include additional error context which
+  includes the conflicted documents meta-data and contents. Note that this error can only be
+  returned if the TrueConflicts parameter is set to true.
+
+- Error:Aborted Code:CAS_MISMATCH
+  This case will occur when a CAS was explicitly provided to the operation, but the target document
+  has been updated and has a different CAS value.
+
+- Error:Aborted Code:DOC_NEWER
+  This case occurs when a new document is rejected because it is older than the document that
+  already exists. The concept of 'newer' uses the buckets configured conflict resolution mode.
+
+ConflictLogging=false:
+
+### CheckDocument
+
+CheckDocument has precisely the same mechanics as PushDocument does, with the exception that in the
+cases where a PushDocument operation would write changes to the document, CheckDocument will return
+success and do no write.

gateway/dataimpl/dataimpl.go

@@ -26,6 +26,7 @@ type Servers struct {
 	AdminQueryIndexV1Server  *server_v1.QueryIndexAdminServer
 	AdminSearchIndexV1Server *server_v1.SearchIndexAdminServer
 	TransactionsV1Server     *server_v1.TransactionsServer
+	XdcrV1Server             *server_v1.XdcrServer
 }
 
 func New(opts *NewOptions) *Servers {
@@ -87,5 +88,10 @@ func New(opts *NewOptions) *Servers {
 			v1ErrHandler,
 			v1AuthHandler,
 		),
+		XdcrV1Server: server_v1.NewXdcrServer(
+			opts.Logger.Named("xdcr"),
+			v1ErrHandler,
+			v1AuthHandler,
+		),
 	}
 }

gateway/dataimpl/server_v1/errorhandler.go

@@ -602,6 +602,17 @@ func (e ErrorHandler) NewDocCasMismatchStatus(baseErr error, bucketName, scopeNa
 	return st
 }
 
+func (e ErrorHandler) NewDocConflictStatus(baseErr error, bucketName, scopeName, collectionName, docId string) *status.Status {
+	st := status.New(codes.Aborted,
+		fmt.Sprintf("Conflict resolution rejected '%s' in '%s/%s/%s'.",
+			docId, bucketName, scopeName, collectionName))
+	st = e.tryAttachStatusDetails(st, &epb.ErrorInfo{
+		Reason: "DOC_NEWER",
+	})
+	st = e.tryAttachExtraContext(st, baseErr)
+	return st
+}
+
 func (e ErrorHandler) NewZeroCasStatus() *status.Status {
 	st := status.New(codes.InvalidArgument, "CAS value cannot be zero.")
 	return st
@@ -1002,3 +1013,21 @@ func (e ErrorHandler) NewInvalidKeyLengthStatus(key string) *status.Status {
 		fmt.Sprintf("Length of document key '%s' must be between 1 and 251 characters.", key))
 	return st
 }
+
+func (e ErrorHandler) NewVbUuidDivergenceStatus(baseErr error, bucketName, scopeName, collectionName, docId string) *status.Status {
+	st := status.New(codes.Aborted,
+		fmt.Sprintf("The specified vbuuid for '%s' in '%s/%s/%s' did not match.",
+			docId, bucketName, scopeName, collectionName))
+	st = e.tryAttachStatusDetails(st, &epb.ErrorInfo{
+		Reason: "VBUUID_MISMATCH",
+	})
+	st = e.tryAttachExtraContext(st, baseErr)
+	return st
+}
+
+func (e ErrorHandler) NewUnimplementedServerVersionStatus() *status.Status {
+	st := status.New(
+		codes.Unimplemented,
+		"The requested feature is not available on this server version.")
+	return st
+}

gateway/dataimpl/server_v1/kvserver.go

@@ -192,7 +192,7 @@ func (s *KvServer) GetAndLock(ctx context.Context, in *kv_v1.GetAndLockRequest)
 	opts.ScopeName = in.ScopeName
 	opts.CollectionName = in.CollectionName
 	opts.Key = []byte(in.Key)
-	opts.LockTime = in.LockTime
+	opts.LockTime = in.LockTimeSecs
 
 	result, err := bucketAgent.GetAndLock(ctx, &opts)
 	if err != nil {