Fix BigTable multi-entry row cached Tail.

logdog/common/storage/bigtable/storage.go

 // Copyright 2015 The LUCI Authors. All rights reserved.
 // Use of this source code is governed under the Apache License, Version 2.0
 // that can be found in the LICENSE file.
 
 package bigtable
 
 import (
         "bytes"
         "errors"
         "fmt"
@@ skipping 208 matching lines @@
         err := s.raw.getLogData(ctx, startKey, r.Limit, r.KeysOnly, func(rk *rowKey, data []byte) error {
                 // Does this key match our requested log stream? If not, we've moved past
                 // this stream's records and must stop iteration.
                 if !rk.sharesPathWith(startKey) {
                         return errStop
                 }
 
                 // Calculate the start index of the contiguous row. Since we index the row
                 // on the LAST entry in the row, count backwards to get the index of the
                 // first entry.
-                startIndex := rk.index - rk.count + 1
+                startIndex := rk.firstIndex()
                 if startIndex < 0 {
                         return storage.ErrBadData
                 }
 
                 // Split our data into records. Leave the records slice nil if we're doing
                 // a keys-only get.
                 var records [][]byte
                 if !r.KeysOnly {
                         var err error
                         if records, err = recordio.Split(data); err != nil {
@@ skipping 76 matching lines @@
 
         // Iterate through all log keys in the stream. Record the latest contiguous
         // one.
         var (
                 rk        = newRowKey(string(project), string(path), startIdx, 0)
                 latest    *rowKey
                 nextIndex = startIdx
         )
         err := s.raw.getLogData(ctx, rk, 0, true, func(rk *rowKey, data []byte) error {
                 // If this record is non-contiguous, we're done iterating.
-                if (rk.index - rk.count + 1) != nextIndex {
+                if rk.firstIndex() != nextIndex {
                         return errStop
                 }
 
                 latest, nextIndex = rk, rk.index+1
                 return nil
         })
         if err != nil && err != errStop {
                 log.Fields{
                         log.ErrorKey: err,
                         "project":    s.Project,
                         "instance":   s.Instance,
                         "table":      s.LogTable,
                 }.Errorf(ctx, "Failed to scan for tail.")
         }
 
         if latest == nil {
                 // No rows for the specified stream.
                 return nil, storage.ErrDoesNotExist
         }
 
         // Update our cache if the tail index has changed.
         if s.Cache != nil && startIdx != latest.index {
-                putLastTailIndex(s, s.Cache, project, path, latest.index)
+                // We cache the first index in the row so that subsequent cached fetches
+                // have the correct "startIdx" expectations.
+                putLastTailIndex(s, s.Cache, project, path, latest.firstIndex())
         }
 
         // Fetch the latest row's data.
         var d []byte
         err = s.raw.getLogData(ctx, latest, 1, false, func(rk *rowKey, data []byte) error {
                 records, err := recordio.Split(data)
                 if err != nil || len(records) == 0 {
                         return storage.ErrBadData
                 }
                 d = records[len(records)-1]
@@ skipping 70 matching lines @@
         }.Debugf(ctx, "Adding entries to BigTable.")
         if err := raw.putLogData(ctx, rk, w.buf.Bytes()); err != nil {
                 return 0, err
         }
 
         // Reset our buffer state.
         w.buf.Reset()
         w.count = 0
         return flushCount, nil
 }