Migrates document metadata from document body to system xattr. On CAS failure, retrieves current doc body and retries migration if _sync property exists. If _sync property is not found, returns doc and sets requiresImport to true
(ctx context.Context, docid string, existingDoc *sgbucket.BucketDocument, opts *sgbucket.MutateInOptions)
| 384 | // Migrates document metadata from document body to system xattr. On CAS failure, retrieves current doc body and retries |
| 385 | // migration if _sync property exists. If _sync property is not found, returns doc and sets requiresImport to true |
| 386 | func (db *DatabaseCollectionWithUser) migrateMetadata(ctx context.Context, docid string, existingDoc *sgbucket.BucketDocument, opts *sgbucket.MutateInOptions) (docOut *Document, err error) { |
| 387 | |
| 388 | // Unmarshal the existing doc in legacy SG format |
| 389 | doc, unmarshalErr := unmarshalDocument(docid, existingDoc.Body) |
| 390 | if unmarshalErr != nil { |
| 391 | return nil, unmarshalErr |
| 392 | } |
| 393 | doc.Cas = existingDoc.Cas |
| 394 | |
| 395 | // If no sync metadata is present, return for import handling |
| 396 | if !doc.HasValidSyncData() { |
| 397 | base.InfofCtx(ctx, base.KeyMigrate, "During migrate, doc %q doesn't have valid sync data.", base.UD(docid)) |
| 398 | return doc, fmt.Errorf("not imported, invalid sync data found") |
| 399 | } |
| 400 | |
| 401 | // Move any large revision bodies to external storage |
| 402 | err = doc.migrateRevisionBodies(ctx, db.dataStore) |
| 403 | if err != nil { |
| 404 | base.InfofCtx(ctx, base.KeyMigrate, "Error migrating revision bodies to external storage, doc %q, (cas=%d), Error: %v", base.UD(docid), doc.Cas, err) |
| 405 | } |
| 406 | |
| 407 | // Persist the document in xattr format |
| 408 | value, syncXattr, vvXattr, _, globalXattr, marshalErr := doc.MarshalWithXattrs() |
| 409 | if marshalErr != nil { |
| 410 | return nil, marshalErr |
| 411 | } |
| 412 | |
| 413 | xattrs := map[string][]byte{ |
| 414 | base.SyncXattrName: syncXattr, |
| 415 | } |
| 416 | if vvXattr != nil { |
| 417 | xattrs[base.VvXattrName] = vvXattr |
| 418 | } |
| 419 | if globalXattr != nil { |
| 420 | xattrs[base.GlobalXattrName] = globalXattr |
| 421 | } |
| 422 | |
| 423 | var casOut uint64 |
| 424 | var writeErr error |
| 425 | var xattrsToDelete []string |
| 426 | if doc.hasFlag(channels.Deleted) { |
| 427 | // Migration of tombstone. Delete body, update xattrs |
| 428 | casOut, writeErr = db.dataStore.WriteTombstoneWithXattrs(ctx, docid, existingDoc.Expiry, existingDoc.Cas, xattrs, xattrsToDelete, true, opts) |
| 429 | } else { |
| 430 | // Non-tombstone - update doc and xattrs |
| 431 | casOut, writeErr = db.dataStore.WriteWithXattrs(ctx, docid, existingDoc.Expiry, existingDoc.Cas, value, xattrs, xattrsToDelete, opts) |
| 432 | } |
| 433 | if writeErr == nil { |
| 434 | doc.Cas = casOut |
| 435 | base.InfofCtx(ctx, base.KeyMigrate, "Successfully migrated doc %q", base.UD(docid)) |
| 436 | return doc, nil |
| 437 | } |
| 438 | |
| 439 | // If it was a cas mismatch, propagate an error as far up the stack as possible to force a full refresh + retry |
| 440 | if base.IsCasMismatch(writeErr) { |
| 441 | return nil, base.ErrCasFailureShouldRetry |
| 442 | } |
| 443 |