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)
| 373 | // nextSequence. |
| 374 | |
| 375 | func (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) |
nothing calls this directly
no test coverage detected