Archivist asserts completeness through keys scan.

server/internal/logdog/archivist/archivist.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 archivist
 
 import (
         "bytes"
         "encoding/hex"
         "fmt"
@@ skipping 154 matching lines @@
         age := ls.Age.Duration()
         if age < at.SettleDelay.Duration() {
                 log.Fields{
                         "age":         age,
                         "settleDelay": at.SettleDelay.Duration(),
                 }.Infof(c, "Log stream is younger than the settle delay. Returning task to queue.")
                 return false, errors.New("log stream is within settle delay")
         }
 
         // Are we required to archive a complete log stream?
-        complete := (age <= at.CompletePeriod.Duration())
-        if complete && ls.State.TerminalIndex < 0 {
-                log.Warningf(c, "Cannot archive complete stream with no terminal index.")
-                return false, errors.New("completeness required, but stream has no terminal index")
+        if age <= at.CompletePeriod.Duration() {
+                tidx := ls.State.TerminalIndex
+
+                if tidx < 0 {
+                        log.Warningf(c, "Cannot archive complete stream with no terminal index.")
+                        return false, errors.New("completeness required, but stream has no terminal index")
+                }
+
+                // If we're requiring completeness, perform a keys-only scan of intermediate
+                // storage to ensure that we have all of the records before we bother
+                // streaming to storage only to find that we are missing data.
+                if err := a.checkComplete(types.StreamPath(at.Path), types.MessageIndex(tidx)); err != nil {
+                        return false, err
+                }
         }
 
         ar := logdog.ArchiveStreamRequest{
                 Path: at.Path,
         }
 
         // Archive to staging.
         //
         // If a non-transient failure occurs here, we will report it to the Archivist
         // under the assumption that it will continue occurring.
         //
         // We will handle error creating the plan and executing the plan in the same
         // switch statement below.
         staged, err := a.makeStagedArchival(c, types.StreamPath(at.Path), ls, task.UniqueID())
         if err != nil {
                 log.WithError(err).Errorf(c, "Failed to create staged archival plan.")
         } else {
-                err = staged.stage(c, complete)
+                err = staged.stage(c)
         }
 
         switch {
         case errors.IsTransient(err):
                 // If this is a transient error, exit immediately and do not delete the
                 // archival task.
                 log.WithError(err).Warningf(c, "TRANSIENT error during archival operation.")
                 return false, err
 
         case err != nil:
@@ skipping 45 matching lines @@
         if _, err := a.Service.ArchiveStream(c, &ar); err != nil {
                 log.WithError(err).Errorf(c, "Failed to report archive state.")
                 return false, err
         }
 
         // Archival is complete and acknowledged by Coordinator. Consume the archival
         // task.
         return true, nil
 }
 
+// checkComplete performs a quick scan of intermediate storage to ensure that
+// all of the log stream's records are available.
+func (a *Archivist) checkComplete(path types.StreamPath, tidx types.MessageIndex) error {
+        sreq := storage.GetRequest{
+                Path:     path,
+                KeysOnly: true,
+        }
+
+        nextIndex := types.MessageIndex(0)
+        var ierr error
+        err := a.Storage.Get(sreq, func(idx types.MessageIndex, d []byte) bool {
+                switch {
+                case idx != nextIndex:
+                        ierr = fmt.Errorf("missing log entry index %d (next %d)", nextIndex, idx)
+                        return false
+
+                case idx == tidx:
+                        // We have hit our terminal index, so all of the log data is here!
+                        return false
+
+                default:
+                        nextIndex++
+                        return true
+                }
+        })
+        if ierr != nil {
+                return ierr
+        }
+        if err != nil {
+                return err
+        }
+        return nil
+}
+
 func (a *Archivist) makeStagedArchival(c context.Context, path types.StreamPath, ls *logdog.LoadStreamResponse, uid string) (
         *stagedArchival, error) {
         sa := stagedArchival{
                 Archivist: a,
                 path:      path,
 
                 terminalIndex: ls.State.TerminalIndex,
         }
 
         // Deserialize and validate the descriptor protobuf. If this fails, it is a
@@ skipping 27 matching lines @@
                 final:  a.GSBase.Concat(string(path), name),
         }
 }
 
 type stagedArchival struct {
         *Archivist
 
         path types.StreamPath
         desc logpb.LogStreamDescriptor
 
-        stream     stagingPaths
-        streamSize int64
-
-        index     stagingPaths
-        indexSize int64
-
-        data     stagingPaths
-        dataSize int64
+        stream stagingPaths
+        index  stagingPaths
+        data   stagingPaths
 
         finalized     bool
         terminalIndex int64
         logEntryCount int64
 }
 
 // stage executes the archival process, archiving to the staged storage paths.
 //
 // If stage fails, it may return a transient error.
-func (sa *stagedArchival) stage(c context.Context, complete bool) (err error) {
+func (sa *stagedArchival) stage(c context.Context) (err error) {
         log.Fields{
                 "streamURL": sa.stream.staged,
                 "indexURL":  sa.index.staged,
                 "dataURL":   sa.data.staged,
         }.Debugf(c, "Staging log stream...")
 
         // Group any transient errors that occur during cleanup. If we aren't
         // returning a non-transient error, return a transient "terr".
         var terr errors.MultiError
         defer func() {
@@ skipping 52 matching lines @@
         if dataWriter, err = createWriter(sa.data.staged); err != nil {
                 return err
         }
         defer closeWriter(dataWriter, sa.data.staged)
 
         // Read our log entries from intermediate storage.
         ss := storageSource{
                 Context:       c,
                 st:            sa.Storage,
                 path:          sa.path,
-                contiguous:    complete,
                 terminalIndex: types.MessageIndex(sa.terminalIndex),
                 lastIndex:     -1,
         }
 
         m := archive.Manifest{
                 Desc:             &sa.desc,
                 Source:           &ss,
                 LogWriter:        streamWriter,
                 IndexWriter:      indexWriter,
                 DataWriter:       dataWriter,
                 StreamIndexRange: sa.StreamIndexRange,
                 PrefixIndexRange: sa.PrefixIndexRange,
                 ByteRange:        sa.ByteRange,
 
                 Logger: log.Get(c),
         }
         if err = archive.Archive(m); err != nil {
                 log.WithError(err).Errorf(c, "Failed to archive log stream.")
                 return
         }
 
-        if tidx := sa.terminalIndex; tidx != int64(ss.lastIndex) {
-                // Fail if we were requested to archive only the complete log. We consider
-                // this a transient error with the expectation that the missing entries will
-                // show up in future retries.
-                switch {
-                case complete && ss.hasMissingEntries:
-                        log.Errorf(c, "Log stream has missing entries, but completeness is required.")
-                        err = errors.WrapTransient(errors.New("stream has missing entries"))
-                        return
+        switch {
+        case ss.logEntryCount == 0:
+                // If our last log index was <0, then no logs were archived.
+                log.Warningf(c, "No log entries were archived.")
 
-                case ss.logEntryCount == 0:
-                        // If our last log index was <0, then no logs were archived.
-                        log.Warningf(c, "No log entries were archived.")
-
-                default:
-                        // Update our terminal index.
-                        log.Fields{
-                                "terminalIndex":     ss.lastIndex,
-                                "logEntryCount":     ss.logEntryCount,
-                                "hasMissingEntries": ss.hasMissingEntries,
-                        }.Debugf(c, "Finished archiving log stream.")
-                }
+        default:
+                // Update our terminal index.
+                log.Fields{
+                        "terminalIndex": ss.lastIndex,
+                        "logEntryCount": ss.logEntryCount,
+                }.Debugf(c, "Finished archiving log stream.")
         }
 
         // Update our state with archival results.
         sa.terminalIndex = int64(ss.lastIndex)
         sa.logEntryCount = ss.logEntryCount
         sa.stream.count = streamWriter.Count()
         sa.index.count = indexWriter.Count()
         sa.data.count = dataWriter.Count()
         return
 }
@@ skipping 66 matching lines @@
         }
 }
 
 func (sa *stagedArchival) getStagingPaths() []*stagingPaths {
         return []*stagingPaths{
                 &sa.stream,
                 &sa.index,
                 &sa.data,
         }
 }