ADDING CHANGES: Note that DocChanged may be executed concurrently for multiple events (in the DCP case, DCP events originating from multiple vbuckets). Only processEntry is locking - all other functionality needs to support concurrent processing.
(event sgbucket.FeedEvent, docType DocumentType)
| 314 | // originating from multiple vbuckets). Only processEntry is locking - all other functionality needs to support |
| 315 | // concurrent processing. |
| 316 | func (c *changeCache) DocChanged(event sgbucket.FeedEvent, docType DocumentType) { |
| 317 | ctx := c.logCtx |
| 318 | docID := string(event.Key) |
| 319 | docJSON := event.Value |
| 320 | changedChannelsCombined := channels.Set{} |
| 321 | |
| 322 | timeReceived := channels.NewFeedTimestamp(&event.TimeReceived) |
| 323 | // ** This method does not directly access any state of c, so it doesn't lock. |
| 324 | // Is this a user/role doc for this database? |
| 325 | switch docType { |
| 326 | case DocTypeUnknown: |
| 327 | return // no-op unknown doc type |
| 328 | case DocTypeUser: |
| 329 | c.processPrincipalDoc(ctx, docID, docJSON, true, timeReceived) |
| 330 | return |
| 331 | case DocTypeRole: |
| 332 | c.processPrincipalDoc(ctx, docID, docJSON, false, timeReceived) |
| 333 | return |
| 334 | case DocTypeUnusedSeq: |
| 335 | c.processUnusedSequence(ctx, docID, timeReceived) |
| 336 | return |
| 337 | case DocTypeUnusedSeqRange: |
| 338 | c.processUnusedSequenceRange(ctx, docID) |
| 339 | return |
| 340 | case DocTypeSGCfg: |
| 341 | if c.cfgEventCallback != nil { |
| 342 | c.cfgEventCallback(docID, event.Cas, nil) |
| 343 | } |
| 344 | return |
| 345 | } |
| 346 | |
| 347 | collection, exists := c.db.CollectionByID[event.CollectionID] |
| 348 | if !exists { |
| 349 | cID := event.CollectionID |
| 350 | if cID == base.DefaultCollectionID && base.MetadataCollectionID == base.DefaultCollectionID { |
| 351 | // It's possible for the `_default` collection to be associated with other databases writing non-principal documents, |
| 352 | // but we still need this collection's feed for the sgCfgPrefix docs. |
| 353 | } else if cID == base.MetadataCollectionID { |
| 354 | // When Metadata moves to a different collection, we should start to warn again - we don't expect non-metadata mutations here! |
| 355 | base.WarnfCtx(ctx, "DocChanged(): Non-metadata mutation for doc %q in MetadataStore - kv ID: %d", base.UD(docID), cID) |
| 356 | } else { |
| 357 | // Unrecognised collection |
| 358 | // we shouldn't be receiving mutations for a collection we're not running a database for (except the metadata store) |
| 359 | base.WarnfCtx(ctx, "DocChanged(): Could not find collection for doc %q - kv ID: %d", base.UD(docID), cID) |
| 360 | } |
| 361 | return |
| 362 | } |
| 363 | |
| 364 | ctx = collection.AddCollectionContext(ctx) |
| 365 | |
| 366 | // If this is a delete and there are no xattrs (no existing SG revision), we can ignore |
| 367 | if event.Opcode == sgbucket.FeedOpDeletion && len(docJSON) == 0 { |
| 368 | base.DebugfCtx(ctx, base.KeyCache, "Ignoring delete mutation for %s - no existing Sync Gateway metadata.", base.UD(docID)) |
| 369 | return |
| 370 | } |
| 371 | |
| 372 | // If this is a binary document (and not one of the above types), we can ignore. Currently only performing this check when xattrs |
| 373 | // are enabled, because walrus doesn't support DataType on feed. |