Don’t hang on diskwriter errors

diskwriter.go

@@ -55,6 +55,7 @@ func NewDiskWriter(ctx context.Context, dest string, opt DiskWriterOpt) (*DiskWr
 		eg:     eg,
 		ctx:    ctx,
 		cancel: cancel,
+		filter: opt.Filter,
 	}, nil
 }
 

receive.go

@@ -106,7 +106,12 @@ func (r *receiver) run(ctx context.Context) error {
 
 	w := newDynamicWalker()
 
-	g.Go(func() error {
+	g.Go(func() (retErr error) {
+		defer func() {
+			if retErr != nil {
+				r.conn.SendMsg(&Packet{Type: PACKET_ERR, Data: []byte(retErr.Error())})
+			}
+		}()
 		destWalker := emptyWalker
 		if !r.merge {
 			destWalker = GetWalkerFn(r.dest)

send.go

@@ -90,6 +90,8 @@ func (s *sender) run(ctx context.Context) error {
 				return err
 			}
 			switch p.Type {
+			case PACKET_ERR:
+				return errors.Errorf("error from receiver: %s", p.Data)
 			case PACKET_REQ:
 				if err := s.queue(p.ID); err != nil {
 					return err

wire.proto

@@ -10,6 +10,7 @@ message Packet {
       PACKET_REQ = 1;
       PACKET_DATA = 2;
       PACKET_FIN = 3;
+      PACKET_ERR = 4;
     }
   PacketType type = 1;
   Stat stat = 2;