Fix BigTable multi-entry row cached Tail.

logdog/common/storage/bigtable/rowKey.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"
         "crypto/sha256"
         "encoding/base64"
@@ skipping 73 matching lines @@
 func (rkb *rowKeyBuffers) value() string {
         return string(rkb.key[:rkb.size])
 }
 
 // rowKey is a BigTable row key.
 //
 // The row key is formed from a Path and its Index. The goal:
 // - Rows with the same path should be clustered.
 // - Rows with the same path should be sorted according to index.
 //
+// The row key index is the index of the LAST entry in the row. Therefore, a
+// row for a given row key will span log indexes [index-count+1..index].
+//
 // Since BigTable rows must be valid UTF8, and since paths are effectively
 // unbounded, the row key will be formed by composing:
 //
 // [ base64(sha256(path)) ] + '~' + [ hex(cmpbin(index)) ] + '~' +
 // [hex(cmpbin(count)]
 type rowKey struct {
         pathHash []byte
         index    int64
         count    int64
 }
@@ skipping 88 matching lines @@
 // than any hex-encoded row index, so this key will always be larger.
 func (rk *rowKey) pathPrefixUpperBound() (v string) {
         withRowKeyBuffers(func(rkb *rowKeyBuffers) {
                 rkb.appendPathPrefix(rk.pathHash)
                 rkb.appendBytes([]byte("~~"))
                 v = rkb.value()
         })
         return
 }
 
+// firstIndex returns the first log entry index represented by this row key.
+func (rk *rowKey) firstIndex() int64 { return rk.index - rk.count + 1 }
+
 // sharesPrefixWith tests if the "path" component of the row key "rk" matches
 // the "path" component of "o".
 func (rk *rowKey) sharesPathWith(o *rowKey) bool {
         return bytes.Equal(rk.pathHash, o.pathHash)
 }
 
 func readHexInt64(v string) (int64, error) {
         d, err := hex.DecodeString(v)
         if err != nil {
                 return 0, errMalformedRowKey
         }
 
         dr := bytes.NewReader(d)
         value, _, err := cmpbin.ReadInt(dr)
         if err != nil {
                 return 0, errMalformedRowKey
         }
 
         // There should be no more data.
         if dr.Len() > 0 {
                 return 0, errMalformedRowKey
         }
 
         return value, nil
 }