Support sending changes for old rev-only docs on a cv changes feed

Author: bbrks

db/changes.go

@@ -42,6 +42,8 @@ type ChangesOptions struct {
 	VersionType    ChangesVersionType // The type of version to use for the changes feed. This is used to determine whether to use send revtree IDs or CV in the changes feed entries.
 }
 
+// ChangesVersionType determines the preferred version type to use in changes feed entries.
+// If the requested version type is not available, the feed will attempt to fall back to ChangesVersionTypeRevTreeID which should always be available.
 type ChangesVersionType string
 
 const (
@@ -60,6 +62,15 @@ func ParseChangesVersionType(s string) (ChangesVersionType, error) {
 	}
 }
 
+// ChangeVersionString attempts to return the version string for the preferred ChangesVersionType, but will fall back to rev if cv is not available when requested.
+func (ce *ChangeEntry) ChangeVersionString(versionType ChangesVersionType) string {
+	if s, ok := ce.Changes[0][versionType]; ok {
+		return s
+	}
+	// requested version type not found, return `rev` as a fallback.
+	return ce.Changes[0][ChangesVersionTypeRevTreeID]
+}
+
 // A changes entry; Database.GetChanges returns an array of these.
 // Marshals into the standard CouchDB _changes format.
 type ChangeEntry struct {
@@ -128,7 +139,10 @@ func (db *DatabaseCollectionWithUser) addDocToChangeEntry(ctx context.Context, e
 			base.WarnfCtx(ctx, "Changes feed: error getting doc %q: %v", base.UD(entry.ID), err)
 			return
 		}
-		db.AddDocInstanceToChangeEntry(ctx, entry, doc, options)
+		if err := db.AddDocInstanceToChangeEntry(ctx, entry, doc, options); err != nil {
+			base.WarnfCtx(ctx, "Changes feed: error adding doc %q to change entry: %v", base.UD(entry.ID), err)
+			return
+		}
 
 	} else if includeConflicts {
 		// Load doc metadata only
@@ -139,11 +153,14 @@ func (db *DatabaseCollectionWithUser) addDocToChangeEntry(ctx context.Context, e
 			base.WarnfCtx(ctx, "Changes feed: error getting doc sync data %q: %v", base.UD(entry.ID), err)
 			return
 		}
-		db.AddDocInstanceToChangeEntry(ctx, entry, doc, options)
+		if err := db.AddDocInstanceToChangeEntry(ctx, entry, doc, options); err != nil {
+			base.WarnfCtx(ctx, "Changes feed: error adding doc %q to change entry: %v", base.UD(entry.ID), err)
+			return
+		}
 
 	} else if options.IncludeDocs {
 		// Retrieve document via rev cache
-		revID := entry.Changes[0]["rev"]
+		revID := entry.ChangeVersionString(options.VersionType)
 		err := db.AddDocToChangeEntryUsingRevCache(ctx, entry, revID)
 		if err != nil {
 			base.WarnfCtx(ctx, "Changes feed: error getting revision body for %q (%s): %v", base.UD(entry.ID), revID, err)
@@ -162,12 +179,16 @@ func (db *DatabaseCollectionWithUser) AddDocToChangeEntryUsingRevCache(ctx conte
 }
 
 // Adds a document body and/or its conflicts to a ChangeEntry
-func (db *DatabaseCollectionWithUser) AddDocInstanceToChangeEntry(ctx context.Context, entry *ChangeEntry, doc *Document, options ChangesOptions) {
+func (db *DatabaseCollectionWithUser) AddDocInstanceToChangeEntry(ctx context.Context, entry *ChangeEntry, doc *Document, options ChangesOptions) error {
 
 	includeConflicts := options.Conflicts && entry.branched
 
-	revID := entry.Changes[0]["rev"]
+	revID := entry.ChangeVersionString(options.VersionType)
 	if includeConflicts {
+		// should've been validated in the handler layer but be defensive
+		if options.VersionType == ChangesVersionTypeCV {
+			return fmt.Errorf("changes feed does not support showing in-conflict revisions when using version_type=cv")
+		}
 		doc.History.forEachLeaf(func(leaf *RevInfo) {
 			if leaf.ID != revID {
 				if !leaf.Deleted {
@@ -187,6 +208,8 @@ func (db *DatabaseCollectionWithUser) AddDocInstanceToChangeEntry(ctx context.Co
 			base.WarnfCtx(ctx, "Changes feed: error getting doc %q/%q: %v", base.UD(doc.ID), revID, err)
 		}
 	}
+
+	return nil
 }
 
 // Parameters
@@ -553,7 +576,7 @@ func makeRevocationChangeEntry(ctx context.Context, logEntry *LogEntry, seqID Se
 }
 
 // AuditReadEvent issues a read event for this change entry. If there is no document body, there will be no event used.
-func (ce *ChangeEntry) AuditReadEvent(ctx context.Context) {
+func (ce *ChangeEntry) AuditReadEvent(ctx context.Context, versionType ChangesVersionType) {
 	if ce.Err != nil {
 		return
 	}
@@ -562,7 +585,7 @@ func (ce *ChangeEntry) AuditReadEvent(ctx context.Context) {
 	}
 	base.Audit(ctx, base.AuditIDDocumentRead, base.AuditFields{
 		base.AuditFieldDocID:      ce.ID,
-		base.AuditFieldDocVersion: ce.Changes[0]["rev"],
+		base.AuditFieldDocVersion: ce.ChangeVersionString(versionType),
 	})
 }
 
@@ -1387,7 +1410,10 @@ func createChangesEntry(ctx context.Context, docid string, db *DatabaseCollectio
 
 	row.Removed = base.SetFromArray(removedChannels)
 	if options.IncludeDocs || options.Conflicts {
-		db.AddDocInstanceToChangeEntry(ctx, row, populatedDoc, options)
+		if err := db.AddDocInstanceToChangeEntry(ctx, row, populatedDoc, options); err != nil {
+			base.WarnfCtx(ctx, "Unable to add doc instance to change entry for %s: %v", base.UD(docid), err)
+			return nil
+		}
 	}
 
 	return row

docs/api/components/schemas.yaml

@@ -549,15 +549,18 @@ Changes-feed:
             type: array
             items:
               type: object
+              minProperties: 1
+              maxProperties: 1
               properties:
                 rev:
-                  description: The new Revision ID that was caused by that change. This is omitted when the `version_type` parameter is `cv`.
+                  description: |
+                    The new Revision ID associated with the change.
+                    This is usually omitted in favour of the `cv` property when the `version_type` preference is set to `cv`,
+                    however data written prior to Sync Gateway 4.0 can still emit `rev` only change entries if we do not have a `cv` available for that document revision.
                   type: string
-                  optional: true
                 cv:
-                  description: The new Current Version that was caused by that change. This is present only when the `version_type` parameter is `cv`.
+                  description: The new Current Version associated with the change. This value requires the `version_type` preference set to `cv`.
                   type: string
-                  optional: true
             uniqueItems: true
       uniqueItems: true
     last_seq:

docs/api/paths/admin/keyspace-_changes.yaml

@@ -104,7 +104,7 @@ get:
         default: false
     - name: version_type
       in: query
-      description: The type of document versioning to use for the changes feed.
+      description: The preferred type of document versioning to use for the changes feed.
       schema:
         type: string
         default: rev
@@ -176,7 +176,7 @@ post:
               type: boolean
               default: false
             version_type:
-              description: The type of document versioning to use for the changes feed.
+              description: The preferred type of document versioning to use for the changes feed.
               type: string
               default: rev
               enum:

rest/changes_api.go

@@ -190,6 +190,10 @@ func (h *handler) handleChanges() error {
 			return base.HTTPErrorf(http.StatusBadRequest, "Invalid version_type: %v", err)
 		}
 
+		if options.Conflicts && options.VersionType == db.ChangesVersionTypeCV {
+			return base.HTTPErrorf(http.StatusBadRequest, "Cannot use 'style=all_docs' with 'version_type=cv'")
+		}
+
 		useRequestPlus, _ := h.getOptBoolQuery("request_plus", h.db.Options.ChangesRequestPlus)
 		if useRequestPlus && feed != feedTypeContinuous {
 			var seqErr error
@@ -424,7 +428,7 @@ func (h *handler) sendSimpleChanges(channels base.Set, options db.ChangesOptions
 					}
 					_ = encoder.Encode(entry)
 					lastSeq = entry.Seq
-					entry.AuditReadEvent(h.ctx())
+					entry.AuditReadEvent(h.ctx(), options.VersionType)
 				}
 
 			case <-heartbeat:
@@ -497,7 +501,7 @@ func (h *handler) sendContinuousChangesByHTTP(inChannels base.Set, options db.Ch
 					break
 				}
 
-				change.AuditReadEvent(h.ctx())
+				change.AuditReadEvent(h.ctx(), options.VersionType)
 			}
 		} else {
 			_, err = h.response.Write([]byte("\n"))
@@ -574,7 +578,7 @@ func (h *handler) sendContinuousChangesByWebSocket(inChannels base.Set, options
 				return err
 			}
 			for _, change := range changes {
-				change.AuditReadEvent(h.ctx())
+				change.AuditReadEvent(h.ctx(), options.VersionType)
 			}
 			return err
 		})

rest/changes_test.go

@@ -571,3 +571,42 @@ func TestChangesVersionType(t *testing.T) {
 		})
 	}
 }
+
+func TestChangesFeedCVWithOldRevOnlyData(t *testing.T) {
+	rt := NewRestTester(t, nil)
+	defer rt.Close()
+
+	seq, err := db.AllocateTestSequence(rt.GetDatabase())
+	require.NoError(t, err)
+	oldDoc := "oldDoc"
+	oldDocBody := []byte(`{"body_field":"1234"}`)
+	oldDocSyncData := []byte(fmt.Sprintf(`{"sequence":%d,"rev":{"rev": "1-abc"},"value_crc32c":"%s"}`, seq, base.Crc32cHashString(oldDocBody)))
+	_, err = rt.GetSingleDataStore().WriteWithXattrs(t.Context(), oldDoc, 0, 0, oldDocBody, map[string][]byte{base.SyncXattrName: oldDocSyncData}, nil, nil)
+	require.NoError(t, err)
+
+	rt.PutDoc("newDoc", `{"foo":"bar"}`)
+
+	rt.WaitForPendingChanges()
+
+	resp := rt.SendAdminRequest(http.MethodGet, "/{{.keyspace}}/_changes?version_type=cv", "")
+	RequireStatus(t, resp, http.StatusOK)
+	var changesResults ChangesResults
+	require.NoError(t, base.JSONUnmarshal(resp.Body.Bytes(), &changesResults))
+	require.Len(t, changesResults.Results, 2)
+	for i, changeEntry := range changesResults.Results {
+		for _, change := range changeEntry.Changes {
+			require.Len(t, change, 1) // ensure only one version type is present
+			// and that it was the expected one (and we have a value)
+			var expectedType db.ChangesVersionType
+			if i == 0 {
+				// first doc was written with a RevID and no CV available
+				expectedType = db.ChangesVersionTypeRevTreeID
+			} else {
+				expectedType = db.ChangesVersionTypeCV
+			}
+			versionValue, ok := change[expectedType]
+			require.Truef(t, ok, "Expected version type %s, got %v", expectedType, change)
+			require.NotEmpty(t, versionValue)
+		}
+	}
+}