Antonio Mika
·
2026-10-10
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 if g.s.c.protocol < 30 && s.opts.recurse {
506 return true
507 }
508 return f.flags&flagContentDir != 0
509 }
510
511 files := make(map[string]bool, len(existing))
512 for _, e := range existing {
513 if name, err := cleanSourcePath(e.Name); err == nil && !e.IsDir {
514 files[name] = true
515 }
516 }
517 scopes := s.filters.newScopeTree(g.dest.dir, g.dest.dir, s.mergeReader(g.dest.dir, files))
518
519 type victim struct {
520 name string
521 isDir bool
522 }
523 var victims []victim
524 var protectedNames []string
525 for _, e := range existing {
526 name, err := cleanSourcePath(e.Name)
527 if err != nil || name == "" {
528 continue
529 }
530 if f, ok := listed[name]; ok && f.isDir() == e.IsDir {
531 continue
532 }
533 // Find the closest listed ancestor. Everything between it and
534 // the entry is unlisted and goes too, unless a filter protects
535 // part of the way.
536 anc := path.Dir(name)
537 var chain []string
538 chain = append(chain, name)
539 for anc != "." {
540 if _, ok := listed[anc]; ok {
541 break
542 }
543 chain = append(chain, anc)
544 anc = path.Dir(anc)
545 }
546 if !contentDir(anc) {
547 continue
548 }
549 // Walk down from the ancestor so merge files are only read in
550 // directories rsync would enter. Below it everything is inside a
551 // directory being deleted.
552 protected := false
553 for i := len(chain) - 1; i >= 0 && !protected; i-- {
554 sc, err := scopes.scope(path.Dir(chain[i]))
555 if err != nil {
556 s.xferError("%v -- skipping file deletion", err)
557 return nil
558 }
559 isDir := i > 0 || e.IsDir
560 if i == len(chain)-1 {
561 protected = s.filters.excluded(chain[i], isDir, sc)
562 } else {
563 protected = s.filters.protectsInDeletedDir(chain[i], isDir, sc)
564 }
565 }
566 if protected {
567 protectedNames = append(protectedNames, name)
568 } else {
569 victims = append(victims, victim{name: name, isDir: e.IsDir})
570 }
571 }
572
573 // Children before parents.
574 slices.SortStableFunc(victims, func(a, b victim) int {
575 return strings.Count(b.name, "/") - strings.Count(a.name, "/")
576 })
577
578 maxDelete := s.opts.maxDelete
579 if maxDelete < 0 && maxDelete != math.MinInt32 {
580 maxDelete = 0
581 }
582 skipped := 0
583 for _, v := range victims {
584 if maxDelete >= 0 && int(g.deletedFiles+g.deletedDirs) >= maxDelete {
585 skipped++
586 continue
587 }
588 if v.isDir && slices.ContainsFunc(protectedNames, func(p string) bool { return strings.HasPrefix(p, v.name+"/") }) {
589 s.info("cannot delete non-empty directory: %s", v.name)
590 continue
591 }
592 full := path.Join(g.dest.dir, v.name)
593 if !s.opts.dryRun {
594 if err := s.fs.Remove(full, v.isDir); err != nil {
595 op := "unlink"
596 if v.isDir {
597 op = "rmdir"
598 }
599 s.xferError("delete_file: %s(%s) failed: %v", op, v.name, err)
600 continue
601 }
602 }
603 if v.isDir {
604 g.deletedDirs++
605 } else {
606 g.deletedFiles++
607 }
608 g.logDelete(v.name, v.isDir)
609 }
610 if skipped > 0 {
611 s.warning("Deletions stopped due to --max-delete limit (%d skipped)", skipped)
612 s.addIOError(ioErrDelLimit)
613 }
614 return nil
615}
616
617func (g *generator) logDelete(name string, isDir bool) {
618 s := g.s
619 if s.c.protocol >= 29 {
620 msg := []byte(name)
621 if isDir {
622 msg = append(msg, 0)
623 }
624 if err := s.c.writeMsg(msgDeleted, msg); err != nil {
625 s.logger.Debug("rsync message not delivered", "err", err)
626 }
627 return
628 }
629 if s.opts.infoDel > 0 {
630 if isDir {
631 name += "/"
632 }
633 s.info("deleting %s", name)
634 }
635}
636
637// receiver applies the deltas the client sends (receiver.c).
638type receiver struct {
639 s *session
640 files []*fileEntry
641 dest destination
642 events chan recvEvent
643 ctx context.Context
644 sent []bool
645}
646
647func (r *receiver) notify(ev recvEvent) error {
648 select {
649 case r.events <- ev:
650 return nil
651 case <-r.ctx.Done():
652 return r.ctx.Err()
653 }
654}
655
656func (r *receiver) run() error {
657 s := r.s
658 c := s.c
659 phase := 0
660 maxPhase := 1
661 if c.protocol >= 29 {
662 maxPhase = 2
663 }
664 tok := newTokenReceiver(c, s.compression)
665 for {
666 req, err := s.readNdxAndAttrs(len(r.files))
667 if err != nil {
668 return err
669 }
670 if req.ndx == ndxDone {
671 phase++
672 if phase > maxPhase {
673 break
674 }
675 if err := r.notify(recvEvent{done: true}); err != nil {
676 return err
677 }
678 continue
679 }
680 f := r.files[req.ndx]
681 if f.inactive {
682 return s.fatal(exitProtocol, fmt.Errorf("refusing transfer of cleared file index %d", req.ndx))
683 }
684 if req.iflags&itemTransfer == 0 {
685 continue
686 }
687 if !f.isReg() {
688 return s.fatal(exitProtocol, fmt.Errorf("received request to transfer non-regular file: %d", req.ndx))
689 }
690 if phase == 2 {
691 return s.fatal(exitProtocol, errors.New("got transfer request in phase 2"))
692 }
693 redoing := r.sent[req.ndx]
694 if s.opts.dryRun {
695 continue
696 }
697
698 err = r.receiveFile(f, tok)
699 switch {
700 case err == nil:
701 if s.opts.removeSource {
702 if err := c.writeMsgInt(msgSuccess, req.ndx); err != nil {
703 return err
704 }
705 }
706 case errors.Is(err, errChecksumMismatch):
707 if redoing {
708 s.xferError("ERROR: %s failed verification -- update discarded.", f.name())
709 continue
710 }
711 if s.opts.infoName > 0 {
712 s.warning("WARNING: %s failed verification -- update discarded (will try again).", f.name())
713 }
714 r.sent[req.ndx] = true
715 if err := r.notify(recvEvent{redo: req.ndx}); err != nil {
716 return err
717 }
718 default:
719 var putErr *putError
720 if !errors.As(err, &putErr) {
721 return err
722 }
723 if errors.Is(putErr.err, ErrFileIgnored) {
724 s.warning("rsync: [receiver] %s skipped: %v", f.name(), putErr.err)
725 } else {
726 s.xferError("%s not stored: %v", f.name(), putErr.err)
727 }
728 }
729 }
730 if err := r.notify(recvEvent{done: true}); err != nil {
731 return err
732 }
733
734 if c.protocol >= 31 {
735 // The sender confirms our final NDX_DONE with one of its own.
736 req, err := s.readNdxAndAttrs(0)
737 if err != nil {
738 return err
739 }
740 if req.ndx != ndxDone {
741 return s.fatal(exitProtocol, fmt.Errorf("invalid packet at end of run (%d)", req.ndx))
742 }
743 return r.notify(recvEvent{done: true})
744 }
745 return nil
746}
747
748// putError is a failure of the storage backend to keep a file.
749type putError struct{ err error }
750
751func (e *putError) Error() string { return e.err.Error() }
752
753// receiveFile reads one file's delta, rebuilds the file from it and our old
754// copy, and streams the result to storage. The stream only ends cleanly
755// once the whole-file checksum matched (receiver.c:receive_data).
756func (r *receiver) receiveFile(f *fileEntry, tok tokenReceiver) error {
757 s := r.s
758 target := r.dest.path(f)
759
760 head, err := s.c.readSumHead(s.cs.xfer.size())
761 if err != nil {
762 return s.fatal(exitProtocol, err)
763 }
764
765 var basis File
766 var basisSize int64
767 if head.count > 0 {
768 b, info, err := s.fs.Open(target)
769 if err == nil {
770 basis, basisSize = b, info.Size
771 defer func() { _ = basis.Close() }()
772 }
773 }
774
775 pr, pw := io.Pipe()
776 type result struct {
777 msg string
778 err error
779 }
780 resc := make(chan result, 1)
781 go func() {
782 info := FileInfo{Name: target, Size: f.size, ModTime: time.Unix(f.mtime, int64(f.nsec))}
783 msg, err := s.fs.Put(target, info, pr)
784 if err == nil {
785 err = errPutReturned
786 }
787 _ = pr.CloseWithError(err)
788 if errors.Is(err, errPutReturned) {
789 err = nil
790 }
791 resc <- result{msg, err}
792 }()
793
794 sum := s.cs.newXferSum()
795 write := func(p []byte) {
796 sum.Write(p)
797 // After a failed Put the pipe returns its error; keep reading the
798 // protocol stream regardless.
799 _, _ = pw.Write(p)
800 }
801
802 var block []byte
803 var tokErr error
804 for {
805 i, data, err := tok.recv()
806 if err != nil {
807 tokErr = err
808 break
809 }
810 if i == 0 {
811 break
812 }
813 if i > 0 {
814 write(data)
815 continue
816 }
817 idx := -(i + 1)
818 if idx >= head.count {
819 tokErr = s.fatal(exitProtocol, fmt.Errorf("invalid block index %d (count=%d)", idx, head.count))
820 break
821 }
822 l := head.blength
823 if idx == head.count-1 && head.remainder != 0 {
824 l = head.remainder
825 }
826 offset := int64(idx) * int64(head.blength)
827 if basis == nil || offset+int64(l) > basisSize {
828 tokErr = s.fatal(exitProtocol, fmt.Errorf("got a block match with no basis file for %q", "/"+target))
829 break
830 }
831 if cap(block) < int(l) {
832 block = make([]byte, l)
833 }
834 block = block[:l]
835 if err := readFullAt(basis, block, offset); err != nil {
836 tokErr = s.fatal(exitFileIO, fmt.Errorf("failed to read basis %q: %w", "/"+target, err))
837 break
838 }
839 tok.see(block)
840 write(block)
841 }
842 if tokErr != nil {
843 _ = pw.CloseWithError(tokErr)
844 <-resc
845 return tokErr
846 }
847
848 remote := make([]byte, s.cs.xfer.size())
849 if err := s.c.readFull(remote); err != nil {
850 _ = pw.CloseWithError(err)
851 <-resc
852 return err
853 }
854 if !bytes.Equal(sum.Sum(nil), remote) {
855 _ = pw.CloseWithError(errChecksumMismatch)
856 <-resc
857 return errChecksumMismatch
858 }
859 _ = pw.Close()
860 res := <-resc
861 if res.err != nil {
862 return &putError{err: res.err}
863 }
864 if res.msg != "" {
865 s.info("%s", res.msg)
866 }
867 return nil
868}
869
870// errPutReturned unblocks writers to the pipe once Put is done reading.
871var errPutReturned = errors.New("rsync: storage finished reading")