Antonio Mika
·
2026-10-14
1package rsync
2
3import (
4 "bytes"
5 "context"
6 "errors"
7 "fmt"
8 "io"
9 "io/fs"
10 "math"
11 "path"
12 "slices"
13 "strings"
14 "time"
15)
16
17var errChecksumMismatch = errors.New("rsync: file failed verification")
18
19// recvEvent is what the receiver tells the generator, standing in for the
20// pipe between rsync's two receiving processes.
21type recvEvent struct {
22 done bool
23 redo int32
24}
25
26// destination maps file-list names to storage paths (get_local_name).
27type destination struct {
28 dir string
29 single bool // the transfer is one file written to dir itself
30}
31
32func (d destination) path(f *fileEntry) string {
33 if d.single {
34 return d.dir
35 }
36 name := f.name()
37 if name == "." {
38 return d.dir
39 }
40 return path.Join(d.dir, name)
41}
42
43// runReceiver is the server side of an upload (do_server_recv).
44func (s *session) runReceiver(args []string) error {
45 if err := s.recvFilterList(); err != nil {
46 return err
47 }
48 files, err := s.recvFileList()
49 if err != nil {
50 return err
51 }
52 if len(files) == 0 {
53 return s.fatal(exitFileSelect, errors.New("server_recv: recv_file_list error"))
54 }
55
56 base := ""
57 if len(args) > 0 {
58 base = args[0]
59 }
60 destArg := ""
61 if len(args) > 1 {
62 destArg = args[1]
63 }
64 dest, err := s.resolveDestination(files, base, destArg)
65 if err != nil {
66 return s.fatal(exitFileSelect, err)
67 }
68
69 ctx, cancel := context.WithCancel(context.Background())
70 defer cancel()
71 events := make(chan recvEvent, 64)
72 g := &generator{s: s, files: files, dest: dest, events: events, ctx: ctx}
73 r := &receiver{s: s, files: files, dest: dest, events: events, ctx: ctx, sent: make([]bool, len(files))}
74
75 errc := make(chan error, 2)
76 go func() { errc <- g.run() }()
77 go func() { errc <- r.run() }()
78 for range 2 {
79 if err := <-errc; err != nil {
80 cancel()
81 return err
82 }
83 }
84 return s.finish()
85}
86
87func (s *session) recvFileList() ([]*fileEntry, error) {
88 codec := s.newFlistCodec()
89 var files []*fileEntry
90 for {
91 xflags, ok, ioErr, err := codec.recvFlags(s.c)
92 if err != nil {
93 return nil, err
94 }
95 if !ok {
96 if !s.opts.ignoreErrors {
97 s.addIOError(ioErr & ioErrMask)
98 }
99 break
100 }
101 f, err := codec.recvEntry(s.c, xflags)
102 if err != nil {
103 return nil, s.fatal(exitProtocol, err)
104 }
105 files = append(files, f)
106 }
107 if !s.opts.numericIDs {
108 for _, want := range []bool{s.opts.preserveUID, s.opts.preserveGID} {
109 if !want {
110 continue
111 }
112 for {
113 id, err := s.c.readVarint30()
114 if err != nil {
115 return nil, err
116 }
117 if id == 0 {
118 break
119 }
120 if _, err := codec.readIDName(s.c); err != nil {
121 return nil, err
122 }
123 }
124 if s.xmitID0Names {
125 if _, err := codec.readIDName(s.c); err != nil {
126 return nil, err
127 }
128 }
129 }
130 }
131 sortFileList(files, s.c.protocol, true)
132 if s.c.protocol < 30 {
133 v, err := s.c.readInt32()
134 if err != nil {
135 return nil, err
136 }
137 if !s.opts.ignoreErrors {
138 s.addIOError(v & ioErrMask)
139 }
140 }
141 return files, nil
142}
143
144func (s *session) resolveDestination(files []*fileEntry, base, destArg string) (destination, error) {
145 dir, err := cleanSourcePath(strings.TrimLeft(path.Join(base, destArg), "/"))
146 if err != nil {
147 return destination{}, err
148 }
149 trailingSlash := strings.HasSuffix(destArg, "/")
150 if dir == "" {
151 return destination{dir: dir}, nil
152 }
153 info, err := s.fs.Stat(dir)
154 exists := err == nil
155 if err != nil && !errors.Is(err, fs.ErrNotExist) {
156 return destination{}, fmt.Errorf("ERROR: cannot stat destination %q: %w", "/"+dir, err)
157 }
158 if exists {
159 if info.IsDir {
160 return destination{dir: dir}, nil
161 }
162 if len(files) > 1 {
163 return destination{}, errors.New("ERROR: destination must be a directory when copying more than 1 file")
164 }
165 if files[0].isDir() {
166 return destination{}, errors.New("ERROR: cannot overwrite non-directory with a directory")
167 }
168 }
169 if len(files) > 1 || trailingSlash {
170 if exists {
171 return destination{}, errors.New("ERROR: destination path is not a directory")
172 }
173 return destination{dir: dir}, nil
174 }
175 return destination{dir: dir, single: true}, nil
176}
177
178// generator decides what the client has to send and sends the signatures
179// of our copies (generator.c).
180type generator struct {
181 s *session
182 files []*fileEntry
183 dest destination
184 events chan recvEvent
185 ctx context.Context
186 done int
187
188 deletedFiles int32
189 deletedDirs int32
190}
191
192func (g *generator) run() error {
193 s := g.s
194 c := s.c
195 earlyDelayDone := !s.opts.delayUpdates
196 earlyDeleteDone := s.opts.deleteDuring != 2 && !s.opts.deleteAfter
197
198 if s.opts.deleteMode && !g.dest.single && (s.opts.deleteBefore || s.opts.deleteDuring == 1) {
199 if err := g.deletePass(); err != nil {
200 return err
201 }
202 }
203
204 for i, f := range g.files {
205 if f.inactive {
206 continue
207 }
208 if err := g.ctx.Err(); err != nil {
209 return err
210 }
211 if err := g.recvGenerator(int32(i), f, false); err != nil {
212 return err
213 }
214 }
215 if err := c.writeNdx(ndxDone); err != nil {
216 return err
217 }
218
219 // Redo files that failed verification until the receiver is done
220 // with the first pass.
221 if err := g.wait(1, true); err != nil {
222 return err
223 }
224
225 if err := c.writeNdx(ndxDone); err != nil {
226 return err
227 }
228 if c.protocol >= 29 && earlyDelayDone {
229 if err := c.writeNdx(ndxDone); err != nil {
230 return err
231 }
232 }
233 if c.protocol >= 31 && earlyDeleteDone {
234 if s.opts.deleteMode {
235 if err := g.writeDelStats(); err != nil {
236 return err
237 }
238 }
239 if earlyDelayDone {
240 if err := c.writeNdx(ndxDone); err != nil {
241 return err
242 }
243 }
244 }
245 if err := g.wait(2, false); err != nil {
246 return err
247 }
248
249 if c.protocol >= 29 {
250 if !earlyDelayDone {
251 if err := c.writeNdx(ndxDone); err != nil {
252 return err
253 }
254 if c.protocol >= 31 && earlyDeleteDone {
255 if err := c.writeNdx(ndxDone); err != nil {
256 return err
257 }
258 }
259 }
260 if err := g.wait(3, false); err != nil {
261 return err
262 }
263 }
264
265 if s.opts.deleteMode && !g.dest.single && (s.opts.deleteDuring == 2 || s.opts.deleteAfter) {
266 if err := g.deletePass(); err != nil {
267 return err
268 }
269 }
270
271 if c.protocol >= 31 {
272 if !earlyDeleteDone {
273 if s.opts.deleteMode {
274 if err := g.writeDelStats(); err != nil {
275 return err
276 }
277 }
278 if err := c.writeNdx(ndxDone); err != nil {
279 return err
280 }
281 }
282 if err := g.wait(4, false); err != nil {
283 return err
284 }
285 }
286
287 // The final goodbye.
288 if err := c.writeNdx(ndxDone); err != nil {
289 return err
290 }
291 return c.flush()
292}
293
294// wait handles receiver events until it has finished n phases.
295func (g *generator) wait(n int, redo bool) error {
296 if err := g.s.c.w.Kick(); err != nil {
297 return err
298 }
299 for g.done < n {
300 select {
301 case <-g.ctx.Done():
302 return g.ctx.Err()
303 case ev := <-g.events:
304 if ev.done {
305 g.done++
306 continue
307 }
308 if redo {
309 if err := g.recvGenerator(ev.redo, g.files[ev.redo], true); err != nil {
310 return err
311 }
312 if err := g.s.c.w.Kick(); err != nil {
313 return err
314 }
315 }
316 }
317 }
318 return nil
319}
320
321// recvGenerator handles one file-list entry (generator.c:recv_generator).
322// A redo asks again for a file that failed verification, with full-length
323// block checksums.
324func (g *generator) recvGenerator(ndx int32, f *fileEntry, redo bool) error {
325 s := g.s
326 o := &s.opts
327 if !f.isReg() {
328 if !f.isDir() && o.infoNonreg > 0 {
329 s.info("skipping non-regular file %q", f.name())
330 }
331 return nil
332 }
333 if !redo {
334 if o.maxSize >= 0 && f.size > o.maxSize {
335 return nil
336 }
337 if o.minSize >= 0 && f.size < o.minSize {
338 return nil
339 }
340 }
341
342 target := g.dest.path(f)
343 st, err := s.fs.Stat(target)
344 exists := err == nil
345 if err != nil && !errors.Is(err, fs.ErrNotExist) {
346 s.xferError("recv_generator: failed to stat %q: %v", "/"+target, err)
347 return nil
348 }
349 if !redo {
350 if o.ignoreNonExist && !exists {
351 return nil
352 }
353 if o.ignoreExisting && exists {
354 return nil
355 }
356 if o.updateOnly && exists && !st.IsDir && f.mtime-st.ModTime.Unix() < 0 {
357 return nil
358 }
359 }
360 if exists && st.IsDir {
361 s.xferError("cannot replace directory %q with a file", "/"+target)
362 return nil
363 }
364
365 if exists && !redo {
366 same, err := g.quickCheck(f, target, st)
367 if err != nil {
368 s.xferError("failed to checksum %q: %v", "/"+target, err)
369 } else if same {
370 if o.removeSource && !o.dryRun {
371 return s.c.writeMsgInt(msgSuccess, ndx)
372 }
373 return nil
374 }
375 }
376
377 // Build the signature of our copy before announcing the file, so a
378 // failure to read it can fall back to sending the whole file.
379 var head sumHead
380 var sums []byte
381 if exists && !o.wholeFile && !o.dryRun && st.Size > 0 {
382 csumLength := shortSumLength
383 if redo {
384 csumLength = sumLength
385 }
386 head, sums, err = g.signature(target, csumLength)
387 if err != nil {
388 s.send(msgError, fmt.Sprintf("rsync: [generator] failed to read %q, continuing: %v\n", "/"+target, err))
389 exists = false
390 head, sums = sumHead{}, nil
391 }
392 }
393
394 if err := s.c.writeNdx(ndx); err != nil {
395 return err
396 }
397 if s.c.protocol >= 29 {
398 iflags := itemTransfer
399 if o.alwaysChecksum {
400 iflags |= itemReportChange
401 }
402 if exists {
403 if st.Size != f.size {
404 iflags |= itemReportSize
405 }
406 if !o.preserveTimes || st.ModTime.Unix() != f.mtime {
407 iflags |= itemReportTime
408 }
409 } else {
410 iflags |= itemIsNew
411 }
412 if err := s.c.writeShortint(uint16(iflags)); err != nil {
413 return err
414 }
415 }
416 if o.dryRun {
417 return nil
418 }
419 if err := s.c.writeSumHead(head); err != nil {
420 return err
421 }
422 return s.c.write(sums)
423}
424
425// signature computes the block checksums of our copy of a file.
426func (g *generator) signature(target string, csumLength int) (sumHead, []byte, error) {
427 s := g.s
428 basis, info, err := s.fs.Open(target)
429 if err != nil {
430 return sumHead{}, nil, err
431 }
432 defer func() { _ = basis.Close() }()
433 if info.Size <= 0 {
434 return sumHead{}, nil, nil
435 }
436 head, err := sumSizesSqroot(info.Size, s.c.protocol, s.opts.blockSize, csumLength, s.cs.xfer.size())
437 if err != nil {
438 return sumHead{}, nil, err
439 }
440 sums, err := encodeSums(basis, info.Size, head, &s.cs)
441 if err != nil {
442 return sumHead{}, nil, err
443 }
444 return head, sums, nil
445}
446
447// quickCheck reports whether our copy already matches (quick_check_ok).
448func (g *generator) quickCheck(f *fileEntry, target string, st FileInfo) (bool, error) {
449 o := &g.s.opts
450 if st.Size != f.size {
451 return false, nil
452 }
453 if o.alwaysChecksum {
454 sum, err := g.s.fileChecksum(target)
455 if err != nil {
456 return false, err
457 }
458 return bytes.Equal(sum, f.sum), nil
459 }
460 if o.sizeOnly {
461 return true, nil
462 }
463 if o.ignoreTimes {
464 return false, nil
465 }
466 // Storage keeps whole seconds, so nanoseconds are not compared.
467 return st.ModTime.Unix() == f.mtime, nil
468}
469
470func (g *generator) writeDelStats() error {
471 return g.s.writeDelStats([5]int32{g.deletedFiles, g.deletedDirs, 0, 0, 0})
472}
473
474// deletePass removes what the destination has beyond the file list.
475func (g *generator) deletePass() error {
476 s := g.s
477 s.mu.Lock()
478 ioError := s.ioError
479 s.mu.Unlock()
480 if ioError != 0 && !s.opts.ignoreErrors {
481 s.info("IO error encountered -- skipping file deletion")
482 return nil
483 }
484
485 existing, err := s.fs.ReadDir(g.dest.dir, true)
486 if errors.Is(err, fs.ErrNotExist) {
487 return nil
488 }
489 if err != nil {
490 s.xferError("delete: failed to list %q: %v", "/"+g.dest.dir, err)
491 return nil
492 }
493
494 listed := make(map[string]*fileEntry, len(g.files))
495 for _, f := range g.files {
496 if !f.inactive {
497 listed[f.name()] = f
498 }
499 }
500 contentDir := func(name string) bool {
501 f, ok := listed[name]
502 if !ok || !f.isDir() {
503 return false
504 }
505 return f.flags&flagContentDir != 0
506 }
507
508 files := make(map[string]bool, len(existing))
509 for _, e := range existing {
510 if name, err := cleanSourcePath(e.Name); err == nil && !e.IsDir {
511 files[name] = true
512 }
513 }
514 scopes := s.filters.newScopeTree(g.dest.dir, g.dest.dir, s.mergeReader(g.dest.dir, files))
515
516 type victim struct {
517 name string
518 isDir bool
519 }
520 var victims []victim
521 var protectedNames []string
522 for _, e := range existing {
523 name, err := cleanSourcePath(e.Name)
524 if err != nil || name == "" {
525 continue
526 }
527 if f, ok := listed[name]; ok && f.isDir() == e.IsDir {
528 continue
529 }
530 // Find the closest listed ancestor. Everything between it and
531 // the entry is unlisted and goes too, unless a filter protects
532 // part of the way.
533 anc := path.Dir(name)
534 var chain []string
535 chain = append(chain, name)
536 for anc != "." {
537 if _, ok := listed[anc]; ok {
538 break
539 }
540 chain = append(chain, anc)
541 anc = path.Dir(anc)
542 }
543 if !contentDir(anc) {
544 continue
545 }
546 // Walk down from the ancestor so merge files are only read in
547 // directories rsync would enter. Below it everything is inside a
548 // directory being deleted.
549 protected := false
550 for i := len(chain) - 1; i >= 0 && !protected; i-- {
551 sc, err := scopes.scope(path.Dir(chain[i]))
552 if err != nil {
553 s.xferError("%v -- skipping file deletion", err)
554 return nil
555 }
556 isDir := i > 0 || e.IsDir
557 if i == len(chain)-1 {
558 protected = s.filters.excluded(chain[i], isDir, sc)
559 } else {
560 protected = s.filters.protectsInDeletedDir(chain[i], isDir, sc)
561 }
562 }
563 if protected {
564 protectedNames = append(protectedNames, name)
565 } else {
566 victims = append(victims, victim{name: name, isDir: e.IsDir})
567 }
568 }
569
570 // Children before parents.
571 slices.SortStableFunc(victims, func(a, b victim) int {
572 return strings.Count(b.name, "/") - strings.Count(a.name, "/")
573 })
574
575 maxDelete := s.opts.maxDelete
576 if maxDelete < 0 && maxDelete != math.MinInt32 {
577 maxDelete = 0
578 }
579 skipped := 0
580 for _, v := range victims {
581 if maxDelete >= 0 && int(g.deletedFiles+g.deletedDirs) >= maxDelete {
582 skipped++
583 continue
584 }
585 if v.isDir && slices.ContainsFunc(protectedNames, func(p string) bool { return strings.HasPrefix(p, v.name+"/") }) {
586 s.info("cannot delete non-empty directory: %s", v.name)
587 continue
588 }
589 full := path.Join(g.dest.dir, v.name)
590 if !s.opts.dryRun {
591 if err := s.fs.Remove(full, v.isDir); err != nil {
592 op := "unlink"
593 if v.isDir {
594 op = "rmdir"
595 }
596 s.xferError("delete_file: %s(%s) failed: %v", op, v.name, err)
597 continue
598 }
599 }
600 if v.isDir {
601 g.deletedDirs++
602 } else {
603 g.deletedFiles++
604 }
605 g.logDelete(v.name, v.isDir)
606 }
607 if skipped > 0 {
608 s.warning("Deletions stopped due to --max-delete limit (%d skipped)", skipped)
609 s.addIOError(ioErrDelLimit)
610 }
611 return nil
612}
613
614func (g *generator) logDelete(name string, isDir bool) {
615 s := g.s
616 if s.c.protocol >= 29 {
617 msg := []byte(name)
618 if isDir {
619 msg = append(msg, 0)
620 }
621 if err := s.c.writeMsg(msgDeleted, msg); err != nil {
622 s.logger.Debug("rsync message not delivered", "err", err)
623 }
624 return
625 }
626 if s.opts.infoDel > 0 {
627 if isDir {
628 name += "/"
629 }
630 s.info("deleting %s", name)
631 }
632}
633
634// receiver applies the deltas the client sends (receiver.c).
635type receiver struct {
636 s *session
637 files []*fileEntry
638 dest destination
639 events chan recvEvent
640 ctx context.Context
641 sent []bool
642}
643
644func (r *receiver) notify(ev recvEvent) error {
645 select {
646 case r.events <- ev:
647 return nil
648 case <-r.ctx.Done():
649 return r.ctx.Err()
650 }
651}
652
653func (r *receiver) run() error {
654 s := r.s
655 c := s.c
656 phase := 0
657 maxPhase := 1
658 if c.protocol >= 29 {
659 maxPhase = 2
660 }
661 tok := newTokenReceiver(c, s.compression)
662 for {
663 req, err := s.readNdxAndAttrs(len(r.files))
664 if err != nil {
665 return err
666 }
667 if req.ndx == ndxDone {
668 phase++
669 if phase > maxPhase {
670 break
671 }
672 if err := r.notify(recvEvent{done: true}); err != nil {
673 return err
674 }
675 continue
676 }
677 f := r.files[req.ndx]
678 if f.inactive {
679 return s.fatal(exitProtocol, fmt.Errorf("refusing transfer of cleared file index %d", req.ndx))
680 }
681 if req.iflags&itemTransfer == 0 {
682 continue
683 }
684 if !f.isReg() {
685 return s.fatal(exitProtocol, fmt.Errorf("received request to transfer non-regular file: %d", req.ndx))
686 }
687 if phase == 2 {
688 return s.fatal(exitProtocol, errors.New("got transfer request in phase 2"))
689 }
690 redoing := r.sent[req.ndx]
691 if s.opts.dryRun {
692 continue
693 }
694
695 err = r.receiveFile(f, tok)
696 switch {
697 case err == nil:
698 if s.opts.removeSource {
699 if err := c.writeMsgInt(msgSuccess, req.ndx); err != nil {
700 return err
701 }
702 }
703 case errors.Is(err, errChecksumMismatch):
704 if redoing {
705 s.xferError("ERROR: %s failed verification -- update discarded.", f.name())
706 continue
707 }
708 if s.opts.infoName > 0 {
709 s.warning("WARNING: %s failed verification -- update discarded (will try again).", f.name())
710 }
711 r.sent[req.ndx] = true
712 if err := r.notify(recvEvent{redo: req.ndx}); err != nil {
713 return err
714 }
715 default:
716 var putErr *putError
717 if !errors.As(err, &putErr) {
718 return err
719 }
720 if errors.Is(putErr.err, ErrFileIgnored) {
721 s.warning("rsync: [receiver] %s skipped: %v", f.name(), putErr.err)
722 } else {
723 s.xferError("%s not stored: %v", f.name(), putErr.err)
724 }
725 }
726 }
727 if err := r.notify(recvEvent{done: true}); err != nil {
728 return err
729 }
730
731 if c.protocol >= 31 {
732 // The sender confirms our final NDX_DONE with one of its own.
733 req, err := s.readNdxAndAttrs(0)
734 if err != nil {
735 return err
736 }
737 if req.ndx != ndxDone {
738 return s.fatal(exitProtocol, fmt.Errorf("invalid packet at end of run (%d)", req.ndx))
739 }
740 return r.notify(recvEvent{done: true})
741 }
742 return nil
743}
744
745// putError is a failure of the storage backend to keep a file.
746type putError struct{ err error }
747
748func (e *putError) Error() string { return e.err.Error() }
749
750// receiveFile reads one file's delta, rebuilds the file from it and our old
751// copy, and streams the result to storage. The stream only ends cleanly
752// once the whole-file checksum matched (receiver.c:receive_data).
753func (r *receiver) receiveFile(f *fileEntry, tok tokenReceiver) error {
754 s := r.s
755 target := r.dest.path(f)
756
757 head, err := s.c.readSumHead(s.cs.xfer.size())
758 if err != nil {
759 return s.fatal(exitProtocol, err)
760 }
761
762 var basis File
763 var basisSize int64
764 if head.count > 0 {
765 b, info, err := s.fs.Open(target)
766 if err == nil {
767 basis, basisSize = b, info.Size
768 defer func() { _ = basis.Close() }()
769 }
770 }
771
772 pr, pw := io.Pipe()
773 type result struct {
774 msg string
775 err error
776 }
777 resc := make(chan result, 1)
778 go func() {
779 info := FileInfo{Name: target, Size: f.size, ModTime: time.Unix(f.mtime, int64(f.nsec))}
780 msg, err := s.fs.Put(target, info, pr)
781 if err == nil {
782 err = errPutReturned
783 }
784 _ = pr.CloseWithError(err)
785 if errors.Is(err, errPutReturned) {
786 err = nil
787 }
788 resc <- result{msg, err}
789 }()
790
791 sum := s.cs.newXferSum()
792 write := func(p []byte) {
793 sum.Write(p)
794 // After a failed Put the pipe returns its error; keep reading the
795 // protocol stream regardless.
796 _, _ = pw.Write(p)
797 }
798
799 var block []byte
800 var tokErr error
801 for {
802 i, data, err := tok.recv()
803 if err != nil {
804 tokErr = err
805 break
806 }
807 if i == 0 {
808 break
809 }
810 if i > 0 {
811 write(data)
812 continue
813 }
814 idx := -(i + 1)
815 if idx >= head.count {
816 tokErr = s.fatal(exitProtocol, fmt.Errorf("invalid block index %d (count=%d)", idx, head.count))
817 break
818 }
819 l := head.blength
820 if idx == head.count-1 && head.remainder != 0 {
821 l = head.remainder
822 }
823 offset := int64(idx) * int64(head.blength)
824 if basis == nil || offset+int64(l) > basisSize {
825 tokErr = s.fatal(exitProtocol, fmt.Errorf("got a block match with no basis file for %q", "/"+target))
826 break
827 }
828 if cap(block) < int(l) {
829 block = make([]byte, l)
830 }
831 block = block[:l]
832 if err := readFullAt(basis, block, offset); err != nil {
833 tokErr = s.fatal(exitFileIO, fmt.Errorf("failed to read basis %q: %w", "/"+target, err))
834 break
835 }
836 tok.see(block)
837 write(block)
838 }
839 if tokErr != nil {
840 _ = pw.CloseWithError(tokErr)
841 <-resc
842 return tokErr
843 }
844
845 remote := make([]byte, s.cs.xfer.size())
846 if err := s.c.readFull(remote); err != nil {
847 _ = pw.CloseWithError(err)
848 <-resc
849 return err
850 }
851 if !bytes.Equal(sum.Sum(nil), remote) {
852 _ = pw.CloseWithError(errChecksumMismatch)
853 <-resc
854 return errChecksumMismatch
855 }
856 _ = pw.Close()
857 res := <-resc
858 if res.err != nil {
859 return &putError{err: res.err}
860 }
861 if res.msg != "" {
862 s.info("%s", res.msg)
863 }
864 return nil
865}
866
867// errPutReturned unblocks writers to the pipe once Put is done reading.
868var errPutReturned = errors.New("rsync: storage finished reading")