Merge pull request #4179 from filecoin-project/fix/export-unclean-exit

chain export: Error with unfinished exports

cli/chain.go

@@ -1070,13 +1070,20 @@ var chainExportCmd = &cli.Command{
 			return err
 		}
 
+		var last bool
 		for b := range stream {
+			last = len(b) == 0
+
 			_, err := fi.Write(b)
 			if err != nil {
 				return err
 			}
 		}
 
+		if !last {
+			return xerrors.Errorf("incomplete export (remote connection lost?)")
+		}
+
 		return nil
 	},
 }

node/impl/full/chain.go

@@ -507,15 +507,11 @@ func (a *ChainAPI) ChainExport(ctx context.Context, nroots abi.ChainEpoch, skipo
 	r, w := io.Pipe()
 	out := make(chan []byte)
 	go func() {
-		defer w.Close() //nolint:errcheck // it is a pipe
-
 		bw := bufio.NewWriterSize(w, 1<<20)
-		defer bw.Flush() //nolint:errcheck // it is a write to a pipe
 
-		if err := a.Chain.Export(ctx, ts, nroots, skipoldmsgs, bw); err != nil {
-			log.Errorf("chain export call failed: %s", err)
-			return
-		}
+		err := a.Chain.Export(ctx, ts, nroots, skipoldmsgs, bw)
+		bw.Flush()            //nolint:errcheck // it is a write to a pipe
+		w.CloseWithError(err) //nolint:errcheck // it is a pipe
 	}()
 
 	go func() {
@@ -527,13 +523,23 @@ func (a *ChainAPI) ChainExport(ctx context.Context, nroots abi.ChainEpoch, skipo
 				log.Errorf("chain export pipe read failed: %s", err)
 				return
 			}
-			select {
-			case out <- buf[:n]:
-			case <-ctx.Done():
-				log.Warnf("export writer failed: %s", ctx.Err())
-				return
+			if n > 0 {
+				select {
+				case out <- buf[:n]:
+				case <-ctx.Done():
+					log.Warnf("export writer failed: %s", ctx.Err())
+					return
+				}
 			}
 			if err == io.EOF {
+				// send empty slice to indicate correct eof
+				select {
+				case out <- []byte{}:
+				case <-ctx.Done():
+					log.Warnf("export writer failed: %s", ctx.Err())
+					return
+				}
+
 				return
 			}
 		}