MCPcopy Create free account
hub / github.com/couchbase/sync_gateway / GetChanges

Method GetChanges

db/channel_cache_single.go:375–457  ·  view source on GitHub ↗

Top-level method to get all the changes in a channel since the sequence 'since'. If the cache doesn't go back far enough, the view will be queried. View query results may be fed back into the cache if there's room. initialSequence is used only if the cache is empty: it gives the max sequence to whic

(ctx context.Context, options ChangesOptions)

Source from the content-addressed store, hash-verified

373// nextSequence.
374
375func (c *singleChannelCacheImpl) GetChanges(ctx context.Context, options ChangesOptions) ([]*LogEntry, error) {
376
377 // Use the cache, and return if it fulfilled the entire request:
378 cacheValidFrom, resultFromCache := c.GetCachedChanges(options)
379 numFromCache := len(resultFromCache)
380 if numFromCache > 0 {
381 base.InfofCtx(ctx, base.KeyCache, "GetCachedChanges(%q, %s) --> %d changes valid from #%d",
382 base.UD(c.channelID), options.Since.String(), numFromCache, cacheValidFrom)
383 } else {
384 base.DebugfCtx(ctx, base.KeyCache, "GetCachedChanges(%q, %s) --> nothing cached",
385 base.UD(c.channelID), options.Since.String())
386 }
387 startSeq := options.Since.SafeSequence() + 1
388 if cacheValidFrom <= startSeq {
389 c.cacheStats.ChannelCacheHits.Add(1)
390 return resultFromCache, nil
391 }
392
393 // Nope, we're going to have to backfill from the view.
394 // ** First acquire the _query_ lock (not the regular lock!)
395 // Track pending queries via StatKeyChannelCachePendingQueries expvar
396 c.cacheStats.ChannelCachePendingQueries.Add(1)
397 c.queryLock.Lock()
398 defer c.queryLock.Unlock()
399 c.cacheStats.ChannelCachePendingQueries.Add(-1)
400
401 // Another goroutine might have gotten the lock first and already queried the view and updated
402 // the cache, so repeat the above:
403 cacheValidFrom, resultFromCache = c.GetCachedChanges(options)
404 if len(resultFromCache) > numFromCache {
405 base.InfofCtx(ctx, base.KeyCache, "2nd GetCachedChanges(%q, %s) got %d more, valid from #%d!",
406 base.UD(c.channelID), options.Since.String(), len(resultFromCache)-numFromCache, cacheValidFrom)
407 }
408 if cacheValidFrom <= startSeq {
409 c.cacheStats.ChannelCacheHits.Add(1)
410 return resultFromCache, nil
411 }
412
413 // Check whether the changes process has been terminated before running a query
414 if options.ChangesCtx.Err() != nil {
415 return nil, fmt.Errorf("Changes feed cancelled %w", options.ChangesCtx.Err())
416 }
417
418 // Now query the view. We set the max sequence equal to cacheValidFrom, so we'll get one
419 // overlap, which helps confirm that we've got everything.
420 c.cacheStats.ChannelCacheMisses.Add(1)
421 endSeq := cacheValidFrom
422 resultFromQuery, err := c.queryHandler.getChangesInChannelFromQuery(ctx, c.channelID.Name, startSeq, endSeq, options.Limit, options.ActiveOnly)
423 if err != nil {
424 return nil, err
425 }
426
427 // Cache some of the query results, if there's room in the cache. If query hit the limit,
428 // the query results are only valid for the range of sequences in the result set.
429 // Don't cache when active_only=true since query results aren't complete.
430 if options.ActiveOnly != true {
431 resultValidTo := endSeq
432 numResults := len(resultFromQuery)

Callers

nothing calls this directly

Calls 13

GetCachedChangesMethod · 0.95
prependChangesMethod · 0.95
InfofCtxFunction · 0.92
UDFunction · 0.92
DebugfCtxFunction · 0.92
SafeSequenceMethod · 0.80
ErrMethod · 0.80
ErrorfMethod · 0.80
StringMethod · 0.65
AddMethod · 0.45
LockMethod · 0.45

Tested by

no test coverage detected