more progress on sync tests

garethgeorge · garethgeorge · Nov 16, 2024 · Nov 29, 2024 · Dec 6, 2024 · Dec 7, 2024
commit 247f4f59728e7f81848b1bf5a449e59c3d630875
diff --git a/internal/api/syncapi/api_test.go → internal/api/syncapi/syncapi_test.go b/internal/api/syncapi/api_test.go → internal/api/syncapi/syncapi_test.go
@@ -224,6 +224,12 @@ func TestSimpleOperationSync(t *testing.T) {
 			DisplayMessage: "hostop1",
 		},
 	})...)
+	peerHost.oplog.Add(testutil.OperationsWithDefaults(basicClientOperationTempl, []*v1.Operation{
+		{
+			DisplayMessage: "clientop-missing",
+			OriginalId:     1234, // must be an ID that doesn't exist remotely
+		},
+	})...)
 
 	peerClient.oplog.Add(testutil.OperationsWithDefaults(basicClientOperationTempl, []*v1.Operation{
 		{
@@ -240,6 +246,19 @@ func TestSimpleOperationSync(t *testing.T) {
 	tryConnect(t, ctx, peerClient, defaultHostID)
 
 	tryExpectOperationsSynced(t, ctx, peerHost, peerClient, oplog.Query{RepoID: defaultRepoID, InstanceID: defaultClientID})
+	tryExpectExactOperations(t, ctx, peerHost, oplog.Query{RepoID: defaultRepoID, InstanceID: defaultClientID},
+		testutil.OperationsWithDefaults(basicClientOperationTempl, []*v1.Operation{
+			{
+				Id:             3, // b/c hostop1 has id 1
+				OriginalId:     1,
+				DisplayMessage: "clientop1",
+			},
+			{
+				Id:             4, // b/c hostop1 has id 1
+				OriginalId:     2,
+				DisplayMessage: "clientop2",
+			},
+		}))
 }
 
 func getOperations(t *testing.T, oplog *oplog.OpLog, query oplog.Query) []*v1.Operation {
@@ -272,6 +291,17 @@ func tryExpectOperationsSynced(t *testing.T, ctx context.Context, peer1 *peerUnd
 	err := testutil.Retry(t, ctx, func() error {
 		peer1Ops := getOperations(t, peer1.oplog, query)
 		peer2Ops := getOperations(t, peer2.oplog, query)
+		// clear fields that we expect will be re-mapped
+		for _, op := range peer1Ops {
+			op.Id = 0
+			op.FlowId = 0
+			op.OriginalId = 0
+		}
+		for _, op := range peer2Ops {
+			op.Id = 0
+			op.FlowId = 0
+			op.OriginalId = 0
+		}
 
 		sortFn := func(a, b *v1.Operation) int {
 			if a.DisplayMessage < b.DisplayMessage {

diff --git a/internal/api/syncapi/syncclient.go b/internal/api/syncapi/syncclient.go
@@ -317,10 +317,10 @@ func (c *SyncClient) runSyncInternal(ctx context.Context) error {
 			}
 		case *v1.SyncStreamItem_DiffOperations:
 			requestedOperations := action.DiffOperations.GetRequestOperations()
+			c.l.Sugar().Debugf("received operation request for operations: %v", requestedOperations)
 
 			var deletedIDs []int64
 			var sendOps []*v1.Operation
-			var ops []*v1.Operation
 
 			sendOpsFunc := func() error {
 				if err := stream.Send(&v1.SyncStreamItem{
@@ -337,6 +337,7 @@ func (c *SyncClient) runSyncInternal(ctx context.Context) error {
 					sendOps = sendOps[:0]
 					return fmt.Errorf("action diff operations: send create operations: %w", err)
 				}
+				c.l.Sugar().Debugf("sent %d operations", len(sendOps))
 				sendOps = sendOps[:0]
 				return nil
 			}
@@ -353,6 +354,7 @@ func (c *SyncClient) runSyncInternal(ctx context.Context) error {
 					continue // skip this operation
 				}
 				if op.GetInstanceId() != c.localInstanceID {
+					c.l.Sugar().Warnf("action diff operations, requested operation %d is not from this instance, this shouldn't happen with a wellbehaved server", opID)
 					continue // skip operations that are not from this instance e.g. an "index snapshot" picking up snapshots created by another instance.
 				}
 
@@ -363,7 +365,7 @@ func (c *SyncClient) runSyncInternal(ctx context.Context) error {
 					return fmt.Errorf("remote requested operation for repo %q for which sync was never initiated", op.GetRepoId())
 				}
 
-				sendOps = append(ops, op)
+				sendOps = append(sendOps, op)
 				sentOps += 1
 				if len(sendOps) >= 256 {
 					if err := sendOpsFunc(); err != nil {

diff --git a/internal/api/syncapi/synchandler.go b/internal/api/syncapi/synchandler.go
@@ -174,9 +174,11 @@ func (h *BackrestSyncHandler) Sync(ctx context.Context, stream *connect.BidiStre
 			// The diff selector _must_ specify a repo the client has access to
 			repo := config.FindRepo(initialConfig, diffSel.GetRepoId())
 			if repo == nil {
+				zap.S().Warnf("syncserver action DiffOperations: client %q tried to diff with repo %q that does not exist", clientInstanceID, diffSel.GetRepoId())
 				return connect.NewError(connect.CodePermissionDenied, fmt.Errorf("action DiffOperations: repo %q not found", diffSel.GetRepoId()))
 			}
 			if !slices.Contains(repo.GetAllowedPeerInstanceIds(), clientInstanceID) {
+				zap.S().Warnf("syncserver action DiffOperations: client %q tried to diff with repo %q that they are not allowed to access", clientInstanceID, diffSel.GetRepoId())
 				return connect.NewError(connect.CodePermissionDenied, fmt.Errorf("action DiffOperations: client is not allowed to access repo %q", diffSel.GetRepoId()))
 			}
 
@@ -192,6 +194,9 @@ func (h *BackrestSyncHandler) Sync(ctx context.Context, stream *connect.BidiStre
 
 			localMetadata := []oplog.OpMetadata{}
 			if err := h.mgr.oplog.QueryMetadata(diffSelQuery, func(metadata oplog.OpMetadata) error {
+				if metadata.OriginalID == 0 {
+					return nil // skip operations that didn't come from a remote
+				}
 				localMetadata = append(localMetadata, metadata)
 				return nil
 			}); err != nil {
@@ -212,6 +217,9 @@ func (h *BackrestSyncHandler) Sync(ctx context.Context, stream *connect.BidiStre
 				return remoteMetadata[i].ID < remoteMetadata[j].ID
 			})
 
+			requestDueToModno := 0
+			requestMissingRemote := 0
+			requestMissingLocal := 0
 			requestIDs := []int64{}
 
 			// This is a simple O(n) diff algorithm that compares the local and remote metadata vectors.
@@ -227,26 +235,39 @@ func (h *BackrestSyncHandler) Sync(ctx context.Context, stream *connect.BidiStre
 					}
 					localIndex++
 					remoteIndex++
+					requestDueToModno++
 				} else if local.OriginalID < remote.ID {
 					// the ID is found locally not remotely, request it and see if we get a delete event back
 					// from the client indicating that the operation was deleted.
 					requestIDs = append(requestIDs, local.OriginalID)
 					localIndex++
+					requestMissingLocal++
 				} else {
 					// the ID is found remotely not locally, request it for initial sync.
 					requestIDs = append(requestIDs, remote.ID)
 					remoteIndex++
+					requestMissingRemote++
 				}
 			}
 			for localIndex < len(localMetadata) {
 				requestIDs = append(requestIDs, localMetadata[localIndex].OriginalID)
 				localIndex++
+				requestMissingLocal++
 			}
 			for remoteIndex < len(remoteMetadata) {
 				requestIDs = append(requestIDs, remoteMetadata[remoteIndex].ID)
 				remoteIndex++
+				requestMissingRemote++
 			}
 
+			zap.L().Debug("syncserver diff operations with client metadata",
+				zap.String("client_instance_id", clientInstanceID),
+				zap.Any("query", diffSelQuery),
+				zap.Int("request_due_to_modno", requestDueToModno),
+				zap.Int("request_local_but_not_remote", requestMissingLocal),
+				zap.Int("request_remote_but_not_local", requestMissingRemote),
+				zap.Int("request_ids_total", len(requestIDs)),
+			)
 			if len(requestIDs) > 0 {
 				if err := stream.Send(&v1.SyncStreamItem{
 					Action: &v1.SyncStreamItem_DiffOperations{
@@ -263,18 +284,21 @@ func (h *BackrestSyncHandler) Sync(ctx context.Context, stream *connect.BidiStre
 		case *v1.SyncStreamItem_SendOperations:
 			switch event := action.SendOperations.GetEvent().Event.(type) {
 			case *v1.OperationEvent_CreatedOperations:
+				zap.L().Debug("syncserver received created operations", zap.Any("operations", event.CreatedOperations.GetOperations()))
 				for _, op := range event.CreatedOperations.GetOperations() {
 					if err := insertOrUpdate(op); err != nil {
 						return fmt.Errorf("action SendOperations: operation event create: %w", err)
 					}
 				}
 			case *v1.OperationEvent_UpdatedOperations:
+				zap.L().Debug("syncserver received update operations", zap.Any("operations", event.UpdatedOperations.GetOperations()))
 				for _, op := range event.UpdatedOperations.GetOperations() {
 					if err := insertOrUpdate(op); err != nil {
 						return fmt.Errorf("action SendOperations: operation event update: %w", err)
 					}
 				}
 			case *v1.OperationEvent_DeletedOperations:
+				zap.L().Debug("syncserver received delete operations", zap.Any("operations", event.DeletedOperations.GetValues()))
 				for _, id := range event.DeletedOperations.GetValues() {
 					if err := deleteByOriginalID(id); err != nil {
 						return fmt.Errorf("action SendOperations: operation event delete %d: %w", id, err)
@@ -285,7 +309,7 @@ func (h *BackrestSyncHandler) Sync(ctx context.Context, stream *connect.BidiStre
 				return connect.NewError(connect.CodeInvalidArgument, errors.New("action SendOperations: unknown event type"))
 			}
 		default:
-			return connect.NewError(connect.CodeInvalidArgument, errors.New("Unknown action type"))
+			return connect.NewError(connect.CodeInvalidArgument, errors.New("unknown action type"))
 		}
 
 		return nil