Method processRev

db/blip_handler.go:1031–1415

Processes a "rev" request, i.e. client is pushing a revision body stats must always be provided, along with all the fields filled with valid pointers

(rq *blip.Message, stats *processRevStats)

// Processes a "rev" request, i.e. client is pushing a revision body
// stats must always be provided, along with all the fields filled with valid pointers
func (bh *blipHandler) processRev(rq *blip.Message, stats *processRevStats) (err error) {
	startTime := time.Now()
	defer func() {
		stats.processingTime.Add(time.Since(startTime).Nanoseconds())
		if err == nil {
			stats.count.Add(1)
		} else {
			stats.errorCount.Add(1)
		}
	}()

	if bh.db.DatabaseContext.Options.UnsupportedOptions != nil && bh.db.DatabaseContext.Options.UnsupportedOptions.RejectWritesWithSkippedSequences {
		// if we are in slow broadcast mode reject write with 503 and increment rejected writes stat
		if bh.db.BroadcastSlowMode.Load() {
			bh.db.DbStats.DatabaseStats.NumDocWritesRejected.Add(1)
			return base.HTTPErrorf(http.StatusServiceUnavailable, "Database cache is behind and cannot accept writes at this time. Please try again later.")
		}
	}

	// throttle concurrent revs
	if cap(bh.inFlightRevsThrottle) > 0 {
		select {
		case bh.inFlightRevsThrottle <- struct{}{}:
		default:
			stats.throttledRevs.Add(1)
			throttleStart := time.Now()
			bh.inFlightRevsThrottle <- struct{}{}
			stats.throttledRevTime.Add(time.Since(throttleStart).Nanoseconds())
		}
		defer func() { <-bh.inFlightRevsThrottle }()
	}

	// addRevisionParams := newAddRevisionParams(rq)
	revMessage := RevMessage{Message: rq}

	// Doc metadata comes from the BLIP message metadata, not magic document properties:
	docID, found := revMessage.ID()
	rev, rfound := revMessage.Rev()
	if !found || !rfound {
		return base.HTTPErrorf(http.StatusBadRequest, "Missing docID or rev")
	}

	if bh.readOnly {
		return base.HTTPErrorf(http.StatusForbidden, "Replication context is read-only, docID: %s, rev:%s", docID, rev)
	}

	base.DebugfCtx(bh.loggingCtx, base.KeySyncMsg, "#%d: Type:%s %s", bh.serialNumber, rq.Profile(), revMessage.String())

	bodyBytes, err := rq.Body()
	if err != nil {
		return err
	}

	base.TracefCtx(bh.loggingCtx, base.KeySyncMsg, "#%d: Properties:%v  Body:%s", bh.serialNumber, base.UD(revMessage.Properties), base.UD(string(bodyBytes)))

	stats.bytes.Add(int64(len(bodyBytes)))

	if bh.BlipSyncContext.purgeOnRemoval && bytes.Contains(bodyBytes, []byte(`"`+BodyRemoved+`":`)) {