LogDog: collector cache non-transient errors.

server/internal/logdog/collector/coordinator/cache.go

 // Copyright 2016 The Chromium Authors. All rights reserved.
 // Use of this source code is governed by a BSD-style license that can be
 // found in the LICENSE file.
 
 package coordinator
 
 import (
         "sync"
         "time"
 
         "github.com/luci/luci-go/common/clock"
         "github.com/luci/luci-go/common/config"
+        "github.com/luci/luci-go/common/errors"
         "github.com/luci/luci-go/common/logdog/types"
         log "github.com/luci/luci-go/common/logging"
         "github.com/luci/luci-go/common/lru"
         "github.com/luci/luci-go/common/promise"
         "github.com/luci/luci-go/common/tsmon/field"
         "github.com/luci/luci-go/common/tsmon/metric"
         "golang.org/x/net/context"
 )
 
 const (
@@ skipping 54 matching lines @@
         key := cacheEntryKey{
                 project: st.Project,
                 path:    st.Path,
         }
 
         // Get the cacheEntry from our cache. If it is expired, doesn't exist, or
         // we're forcing, ignore any existing entry and replace with a Promise pending
         // Coordinator sync.
         cacheHit := false
         entry := c.lru.Mutate(key, func(current interface{}) interface{} {
-                // Don't replace an existing entry, unless it has an error or has expired.
+                // Don't replace an existing entry, unless it has a transient error or has
+                // expired.
                 if current != nil {
                         curEntry := current.(*cacheEntry)
-                        if !curEntry.hasError() && now.Before(curEntry.expiresAt) {
+                        if !curEntry.hasTransientError() && now.Before(curEntry.expiresAt) {
                                 cacheHit = true
                                 return current
                         }
                 }
 
                 p := promise.New(func() (interface{}, error) {
                         st, err := c.Coordinator.RegisterStream(ctx, st, desc)
-                        if err != nil {
-                                return nil, err
-                        }
-                        return st, nil
+                        return st, err
                 })
 
                 return &cacheEntry{
                         cacheEntryKey: cacheEntryKey{
                                 project: st.Project,
                                 path:    st.Path,
                         },
                         terminalIndex: -1,
                         p:             p,
                         expiresAt:     now.Add(c.expiration),
                 }
         }).(*cacheEntry)
 
-        // If there was an error, purge the erroneous entry from the cache so that
-        // the next "update" will re-fetch it.
+        // If there was a transient error, purge the erroneous entry from the cache so
+        // that the next "update" will re-fetch it.
+        //
+        // If the error was non-transient, we will retain it as the Promise return
+        // value and subsequent calls will also receive this error.
         st, err := entry.get(ctx)
-        if err != nil {
+        if errors.IsTransient(err) {
                 log.Fields{
                         log.ErrorKey: err,
-                }.Errorf(ctx, "Error retrieving stream state.")
+                }.Errorf(ctx, "Transient error retrieving stream state.")
                 return nil, err
         }
 
         tsCache.Add(ctx, 1, cacheHit)
-        return st, nil
+        return st, err
 }
 
 func (c *cache) TerminateStream(ctx context.Context, st *LogStreamState) error {
         key := cacheEntryKey{
                 project: st.Project,
                 path:    st.Path,
         }
 
         // Immediately update our state cache to record the terminal index, if
         // we have a state cache.
@@ skipping 61 matching lines @@
         defer e.Unlock()
 
         rp := *(promisedSt.(*LogStreamState))
         if rp.TerminalIndex < 0 {
                 rp.TerminalIndex = e.terminalIndex
         }
 
         return &rp, nil
 }
 
-// hasError tests if this entry has completed evaluation with an error state.
+// hasTransientError tests if this entry has completed evaluation with an error state.
 // This is non-blocking, so if the evaluation hasn't completed, it will return
 // false.
-func (e *cacheEntry) hasError() bool {
-        if _, err := e.p.Peek(); err != nil && err != promise.ErrNoData {
-                return true
+func (e *cacheEntry) hasTransientError() bool {
+        switch _, err := e.p.Peek(); err {
+        case nil, promise.ErrNoData:
+                return false
+        default:
+                return errors.IsTransient(err)
         }
-        return false
 }
 
 // loadTerminalIndex loads a local cache of the stream's terminal index. This
 // will be applied to all future get requests.
 func (e *cacheEntry) loadTerminalIndex(tidx types.MessageIndex) {
         e.Lock()
         defer e.Unlock()
 
         e.terminalIndex = tidx
 }