Fix sparse index handling, add index params.

logdog/common/storage/archive/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 archive implements a storage.Storage instance that retrieves logs
 // from a Google Storage archive.
 //
 // This is a special implementation of storage.Storage, and does not fully
 // conform to the API expecations. Namely:
 //      - It is read-only. Mutation methods will return storage.ErrReadOnly.
@@ skipping 97 matching lines @@
 func (s *storageImpl) Put(storage.PutRequest) error { return storage.ErrReadOnly }
 
 func (s *storageImpl) Get(req storage.GetRequest, cb storage.GetCallback) error {
         idx, err := s.getIndex()
         if err != nil {
                 return err
         }
 
         // Identify the byte offsets that we want to fetch from the entries stream.
         st := s.buildGetStrategy(&req, idx)
-        if st.lastIndex == -1 || req.Index > st.lastIndex {
-                // No records to read.
+        if st.lastIndex >= 0 && req.Index > st.lastIndex {
+                // We know the last index, and the user requested logs past it, so there are
+                // no records to read.
                 return nil
         }
 
         offset := int64(st.startOffset)
         log.Fields{
                 "offset": offset,
                 "length": st.length(),
                 "path":   s.streamPath,
         }.Debugf(s, "Creating stream reader for range.")
         r, err := s.Client.NewReader(s.streamPath, offset, st.length())
@@ skipping 175 matching lines @@
 
 // setEndOffset sets the `length` field. If called multiple times, the smallest
 // assigned value will be retained.
 func (gs *getStrategy) setEndOffset(v uint64) {
         if gs.endOffset == 0 || gs.endOffset > v {
                 gs.endOffset = v
         }
 }
 
 func (s *storageImpl) buildGetStrategy(req *storage.GetRequest, idx *logpb.LogIndex) *getStrategy {
-        st := getStrategy{}
+        st := getStrategy{
+                lastIndex: -1,
+        }
 
         if len(idx.Entries) == 0 {
-                st.lastIndex = -1
                 return &st
         }
 
-        st.lastIndex = types.MessageIndex(idx.Entries[len(idx.Entries)-1].StreamIndex)

[dnj, 2016/10/17 22:12:12]

The actual bug: if there is a sparse index whose last entry, X, ends before the
end of the stream, for a stream with N (>X) entries, and an offset, >X, is
requested, this will falsely assume that the index is contiguous and will say,
"hey, X is the last entry, so there are no logs past X", even though there's
X+1..N.

Now, we only do this optimization if the index contains the cool new "last
stream index" entry.

+        // If we have a log entry count, mark the last log index.
+        if idx.LogEntryCount > 0 {
+                st.lastIndex = types.MessageIndex(idx.Entries[len(idx.Entries)-1].StreamIndex)
+        }
+
         startIdx := indexEntryFor(idx.Entries, req.Index)
         if startIdx < 0 {
                 startIdx = 0
         }
         le := idx.Entries[startIdx]
         st.startOffset = le.Offset
 
         // Determine an upper bound based on our limits.
         //
         // If we have a count limit, and we have enough index entries to upper-bound
@@ skipping 42 matching lines @@
         ui := uint64(i)
         s := sort.Search(len(entries), func(i int) bool {
                 return entries[i].StreamIndex > ui
         })
 
         // The returned index is the one immediately after the index that we want. If
         // our search returned 0, the first index entry is > our search entry, and we
         // will return nil.
         return s - 1
 }