LogDog: Handle archive failures.

server/internal/logdog/archivist/archivist.go

Index: server/internal/logdog/archivist/archivist.go
diff --git a/server/internal/logdog/archivist/archivist.go b/server/internal/logdog/archivist/archivist.go
index 1d79af1d0225532985066c380010d92f0ad693f4..e1cb4587a19126bd48ee7c7d64992391f46d8d88 100644
--- a/server/internal/logdog/archivist/archivist.go
+++ b/server/internal/logdog/archivist/archivist.go
@@ -116,7 +116,18 @@ func (a *Archivist) Archive(c context.Context, t *logdog.ArchiveTask) error {
 	}
 	if err := task.archive(c); err != nil {
 		log.WithError(err).Errorf(c, "Failed to perform archival operation.")
-		return err
+
+		// Fail only if completeness is a requirement. Allow the Coordinator to
+		// handle errors on !Complete streams.
+		if t.Complete {
+			return err
+		}
+
+		// Report that there was an archival error to the Coordinator's
+		// ArchiveStream endpoint. It may accept the archival or respond with a
+		// FailedPrecondition error code indicating that this archival run should
+		// be considered a failure.
+		task.ar.Error = true

[Vadim Sh., 2016/03/31 22:29:59]

are transient errors possible in "task.archive"? Is coordinator involved in
handling them (better to retry locally a bunch of times before bothering
coordinator).

[dnj, 2016/04/01 22:57:04]

They are retried in task.archive already as part of BigTable and GS shims.

 	}
 	log.Fields{
 		"streamURL":     task.ar.StreamUrl,
@@ -124,6 +135,7 @@ func (a *Archivist) Archive(c context.Context, t *logdog.ArchiveTask) error {
 		"dataURL":       task.ar.DataUrl,
 		"terminalIndex": task.ar.TerminalIndex,
 		"complete":      task.ar.Complete,
+		"hadError":      task.ar.Error,
 	}.Debugf(c, "Finished archive construction.")
 
 	if _, err := a.Service.ArchiveStream(c, &task.ar); err != nil {
@@ -151,6 +163,10 @@ type archiveTask struct {
 // Coordinator State. Upon success, the State will be updated with the result
 // of the archival operation.
 func (t *archiveTask) archive(c context.Context) (err error) {
+	// Minimal ArchiveRequest parameters.
+	t.ar.Path = t.Path
+	t.ar.TerminalIndex = -1
+
 	// Generate our archival object managers.
 	bext := t.desc.BinaryFileExt
 	if bext == "" {
@@ -271,7 +287,6 @@ func (t *archiveTask) archive(c context.Context) (err error) {
 	}
 
 	// Update our state with archival results.
-	t.ar.Path = t.Path
 	t.ar.StreamSize = streamO.Count()
 	t.ar.IndexSize = indexO.Count()
 	t.ar.DataSize = dataO.Count()