Antonio Mika
·
2026-10-14
1package rsync
2
3import (
4 "crypto/rand"
5 "encoding/binary"
6 "errors"
7 "fmt"
8 "io"
9 "log/slog"
10 "slices"
11 "strings"
12 "sync"
13
14 "github.com/picosh/pico/pkg/rsync/rsyncopts"
15)
16
17// Config configures a server session.
18type Config struct {
19 FS FS
20 Logger *slog.Logger
21 // Stderr receives errors raised before the protocol is running.
22 Stderr io.Writer
23}
24
25// ExitError carries the exit code the client should see. Its message has
26// already been delivered to the client.
27type ExitError struct {
28 Code int
29 Err error
30}
31
32func (e *ExitError) Error() string {
33 if e.Err == nil {
34 return fmt.Sprintf("rsync exited with code %d", e.Code)
35 }
36 return e.Err.Error()
37}
38
39func (e *ExitError) Unwrap() error { return e.Err }
40
41// options holds the parsed command line in the form the session uses.
42type options struct {
43 sender bool
44 recurse bool
45 relative bool
46 xferDirs bool
47 dryRun bool
48 preserveUID bool
49 preserveGID bool
50 preserveLinks bool
51 preserveDevs bool
52 preserveSpecs bool
53 preserveTimes bool
54 preserveAtimes bool
55 numericIDs bool
56 alwaysChecksum bool
57 ignoreTimes bool
58 sizeOnly bool
59 updateOnly bool
60 ignoreExisting bool
61 ignoreNonExist bool
62 wholeFile bool
63 inplace bool
64 delayUpdates bool
65 removeSource bool
66 deleteMode bool
67 deleteBefore bool
68 deleteDuring int
69 deleteAfter bool
70 deleteExcluded bool
71 pruneEmptyDirs bool
72 ignoreErrors bool
73 cvsExclude bool
74 maxDelete int
75 maxSize int64
76 minSize int64
77 blockSize int32
78 infoName int
79 infoDel int
80 infoNonreg int
81}
82
83type stats struct {
84 literal int64
85 matched int64
86}
87
88// session is one rsync server run.
89type session struct {
90 fs FS
91 logger *slog.Logger
92 c *conn
93 opts options
94
95 cs checksums
96 compression compression
97 compressLevel int
98 filters *filterList
99 varintFlags bool
100 safeFlist bool
101 xmitID0Names bool
102 inplacePartial bool
103
104 mu sync.Mutex
105 ioError int32
106 gotXferError bool
107 stats stats
108}
109
110// Serve runs the server side of an rsync transfer over rw, which carries the
111// SSH session's stdin and stdout. args is the command line after "rsync".
112func Serve(cfg Config, rw io.ReadWriter, args []string) error {
113 logger := cfg.Logger
114 if logger == nil {
115 logger = slog.New(slog.DiscardHandler)
116 }
117 stderr := cfg.Stderr
118 if stderr == nil {
119 stderr = io.Discard
120 }
121 early := func(code int, err error) error {
122 _, _ = fmt.Fprintf(stderr, "rsync: %v\n", err)
123 return &ExitError{Code: code, Err: err}
124 }
125
126 c := newConn(rw)
127 defer func() { _ = c.close() }()
128 pc, err := rsyncopts.ParseArguments(args)
129 if err != nil {
130 return early(exitSyntax, err)
131 }
132 if pc.Options.ProtectArgs() {
133 // The real arguments follow on stdin, NUL-separated.
134 extra, err := readProtectedArgs(c)
135 if err != nil {
136 return early(exitProtocol, err)
137 }
138 if pc, err = rsyncopts.ParseArguments(append(slices.Clone(args), extra...)); err != nil {
139 return early(exitSyntax, err)
140 }
141 }
142 o := pc.Options
143 if !o.Server() {
144 return early(exitSyntax, errors.New("only server mode is supported"))
145 }
146 if err := checkSupported(o); err != nil {
147 return early(exitUnsupported, err)
148 }
149
150 s := &session{
151 fs: cfg.FS,
152 logger: logger,
153 c: c,
154 filters: &filterList{},
155 }
156 s.opts = options{
157 sender: o.Sender(),
158 recurse: o.Recurse(),
159 relative: o.RelativePaths(),
160 xferDirs: o.XferDirs(),
161 dryRun: o.DryRun(),
162 preserveUID: o.PreserveUid(),
163 preserveGID: o.PreserveGid(),
164 preserveLinks: o.PreserveLinks(),
165 preserveDevs: o.PreserveDevices(),
166 preserveSpecs: o.PreserveSpecials(),
167 preserveTimes: o.PreserveMTimes(),
168 preserveAtimes: o.PreserveAtimes(),
169 numericIDs: o.NumericIDs(),
170 alwaysChecksum: o.AlwaysChecksum(),
171 ignoreTimes: o.IgnoreTimes(),
172 sizeOnly: o.SizeOnly(),
173 updateOnly: o.UpdateOnly(),
174 ignoreExisting: o.IgnoreExisting(),
175 ignoreNonExist: o.IgnoreNonExisting(),
176 wholeFile: o.WholeFile(),
177 inplace: o.Inplace(),
178 delayUpdates: o.DelayUpdates(),
179 removeSource: o.RemoveSourceFiles() != 0,
180 deleteMode: o.DeleteMode(),
181 deleteBefore: o.DeleteBefore(),
182 deleteDuring: o.DeleteDuring(),
183 deleteAfter: o.DeleteAfter(),
184 deleteExcluded: o.DeleteExcluded(),
185 pruneEmptyDirs: o.PruneEmptyDirs(),
186 ignoreErrors: o.IgnoreErrors(),
187 cvsExclude: o.CvsExclude(),
188 maxDelete: o.MaxDelete(),
189 maxSize: o.MaxSize(),
190 minSize: o.MinSize(),
191 blockSize: int32(o.BlockSize()),
192 infoName: o.Info(rsyncopts.INFO_NAME),
193 infoDel: o.Info(rsyncopts.INFO_DEL),
194 infoNonreg: o.Info(rsyncopts.INFO_NONREG),
195 }
196 s.filters.deleteExcluded = s.opts.deleteExcluded
197 s.filters.otherSide = ruleSenderSide
198 if s.opts.sender {
199 s.filters.otherSide = ruleReceiverSide
200 }
201
202 if err := s.setupProtocol(o); err != nil {
203 var exit *ExitError
204 if errors.As(err, &exit) {
205 return early(exit.Code, exit.Err)
206 }
207 return early(exitProtocol, err)
208 }
209 s.filters.cvsPerishable = s.c.protocol >= 30
210
211 if err := c.w.setMultiplex(true); err != nil {
212 return &ExitError{Code: exitStreamIO, Err: err}
213 }
214 if s.c.protocol >= 30 {
215 c.r.multiplex = true
216 }
217 c.r.onMsg = s.handleMsg
218
219 logger.Debug("rsync session starting",
220 "protocol", c.protocol,
221 "sender", s.opts.sender,
222 "checksum", s.cs.xfer,
223 "compression", s.compression,
224 )
225
226 if s.opts.sender {
227 err = s.runSender(pc.RemainingArgs)
228 } else {
229 err = s.runReceiver(pc.RemainingArgs)
230 }
231 if err == nil {
232 return nil
233 }
234 var exit *ExitError
235 if errors.As(err, &exit) {
236 return exit
237 }
238 // An unexpected failure, most likely the connection going away.
239 return s.fatal(exitStreamIO, err)
240}
241
242// checkSupported refuses options that would change the wire format in ways
243// we don't implement, or that make no sense for a storage backend.
244func checkSupported(o *rsyncopts.Options) error {
245 switch {
246 case o.PreserveHardLinks():
247 return errors.New("--hard-links is not supported")
248 case o.PreserveACLs():
249 return errors.New("--acls is not supported")
250 case o.PreserveXattrs():
251 return errors.New("--xattrs is not supported")
252 case o.PreserveCrtimes():
253 return errors.New("--crtimes is not supported")
254 case o.Sender() && o.RelativePaths():
255 return errors.New("--relative is not supported when downloading")
256 case o.AppendMode() != 0:
257 return errors.New("--append is not supported")
258 case len(o.BasisDirs()) > 0:
259 return errors.New("--link-dest, --copy-dest and --compare-dest are not supported")
260 case o.BatchMode():
261 return errors.New("batch mode is not supported")
262 case o.FilesFrom() != "":
263 return errors.New("--files-from is not supported")
264 case o.Iconv() != "":
265 return errors.New("--iconv is not supported")
266 case o.MissingArgs() == 2:
267 return errors.New("--delete-missing-args is not supported")
268 case o.Sender() && o.RemoveSourceFiles() != 0:
269 return errors.New("--remove-source-files is not supported when downloading")
270 }
271 return nil
272}
273
274// readProtectedArgs reads the arguments a --secluded-args client sends
275// before the protocol starts. The first one is the program name.
276func readProtectedArgs(c *conn) ([]string, error) {
277 var args []string
278 var cur []byte
279 for {
280 b, err := c.readByte()
281 if err != nil {
282 return nil, err
283 }
284 if b != 0 {
285 cur = append(cur, b)
286 continue
287 }
288 if len(cur) == 0 {
289 break
290 }
291 args = append(args, string(cur))
292 cur = cur[:0]
293 }
294 if len(args) > 0 {
295 args = args[1:]
296 }
297 return args, nil
298}
299
300// setupProtocol runs the handshake that precedes the transfer
301// (compat.c:setup_protocol).
302func (s *session) setupProtocol(o *rsyncopts.Options) error {
303 c := s.c
304 version := maxProtocol
305 if v := o.ProtocolVersion(); v > 0 && v < version {
306 version = v
307 }
308 if err := c.writeInt32(int32(version)); err != nil {
309 return err
310 }
311 if err := c.flush(); err != nil {
312 return err
313 }
314 remote, err := c.readInt32()
315 if err != nil {
316 return err
317 }
318 if int(remote) < version {
319 version = int(remote)
320 }
321 if version < minProtocol {
322 return &ExitError{Code: exitProtocol, Err: fmt.Errorf("protocol version %d is too old, %d or newer is required", remote, minProtocol)}
323 }
324 c.protocol = version
325
326 if s.opts.deleteMode && !s.opts.deleteBefore && s.opts.deleteDuring == 0 && !s.opts.deleteAfter {
327 if version < 30 {
328 s.opts.deleteBefore = true
329 } else {
330 s.opts.deleteDuring = 1
331 }
332 }
333
334 negotiate := false
335 if version >= 30 {
336 info := clientInfo(o.ShellCmd())
337 flags := 0
338 if strings.Contains(info, "f") {
339 flags |= cfSafeFlist
340 }
341 if strings.Contains(info, "x") {
342 flags |= cfAvoidXattrOptim
343 }
344 if strings.Contains(info, "C") {
345 flags |= cfChksumSeedFix
346 }
347 if strings.Contains(info, "I") {
348 flags |= cfInplacePartialDir
349 }
350 if strings.Contains(info, "u") {
351 flags |= cfID0Names
352 }
353 if strings.Contains(info, "v") {
354 negotiate = true
355 flags |= cfVarintFlistFlags
356 }
357 if strings.Contains(info, "V") {
358 flags |= cfVarintFlistFlags
359 err = c.writeByte(byte(flags))
360 } else {
361 err = c.writeVarint(int32(flags))
362 }
363 if err != nil {
364 return err
365 }
366 s.cs.properSeedOrder = flags&cfChksumSeedFix != 0
367 s.varintFlags = flags&cfVarintFlistFlags != 0
368 s.xmitID0Names = flags&cfID0Names != 0
369 s.safeFlist = flags&cfSafeFlist != 0 || version >= 31
370 s.inplacePartial = flags&cfInplacePartialDir != 0
371 }
372
373 checksumChoice := o.ChecksumChoice()
374 compressChoice := o.CompressChoice()
375 wantCompress := o.Compress() || compressChoice != ""
376 if negotiate {
377 if checksumChoice == "" {
378 if err := c.writeVstring(nameList(checksumNames)); err != nil {
379 return err
380 }
381 }
382 if wantCompress && compressChoice == "" {
383 if err := c.writeVstring(nameList(compressNames)); err != nil {
384 return err
385 }
386 }
387 if err := c.flush(); err != nil {
388 return err
389 }
390 }
391
392 // Pick the transfer checksum.
393 switch {
394 case checksumChoice != "":
395 xfer, file, _ := strings.Cut(checksumChoice, ",")
396 if file == "" {
397 file = xfer
398 }
399 if s.cs.xfer, err = checksumByChoice(xfer, version); err != nil {
400 return &ExitError{Code: exitUnsupported, Err: err}
401 }
402 if s.cs.file, err = checksumByChoice(file, version); err != nil {
403 return &ExitError{Code: exitUnsupported, Err: err}
404 }
405 case negotiate:
406 list, err := c.readVstring(256)
407 if err != nil {
408 return err
409 }
410 name, ok := negotiateName(checksumNames, list)
411 if !ok {
412 return &ExitError{Code: exitUnsupported, Err: fmt.Errorf("failed to negotiate a checksum choice, client list: %s", list)}
413 }
414 s.cs.xfer, _ = parseChecksumName(name)
415 s.cs.file = s.cs.xfer
416 default:
417 s.cs.xfer, _ = checksumByChoice("auto", version)
418 s.cs.file = s.cs.xfer
419 }
420
421 // Pick the compression.
422 s.compression = compressNone
423 switch {
424 case compressChoice != "":
425 if s.compression, err = parseCompressName(compressChoice); err != nil {
426 return &ExitError{Code: exitUnsupported, Err: err}
427 }
428 case wantCompress && negotiate:
429 list, err := c.readVstring(256)
430 if err != nil {
431 return err
432 }
433 name, ok := negotiateName(compressNames, list)
434 if !ok {
435 return &ExitError{Code: exitUnsupported, Err: fmt.Errorf("failed to negotiate a compress choice, client list: %s", list)}
436 }
437 s.compression, _ = parseCompressName(name)
438 case wantCompress:
439 s.compression = compressZlib
440 }
441 s.compressLevel = 6
442 if lvl := o.CompressLevel(); s.compression != compressNone && lvl >= 0 {
443 if lvl == 0 {
444 s.compression = compressNone
445 } else {
446 s.compressLevel = min(lvl, 9)
447 }
448 }
449
450 seed := int32(o.ChecksumSeed())
451 if seed == 0 {
452 var b [4]byte
453 _, _ = rand.Read(b[:])
454 seed = int32(binary.LittleEndian.Uint32(b[:]) & 0x7FFFFFFF)
455 if seed == 0 {
456 seed = 1
457 }
458 }
459 s.cs.seed = seed
460 if err := c.writeInt32(seed); err != nil {
461 return err
462 }
463 if s.cs.xfer == csumNone {
464 s.opts.wholeFile = true
465 }
466 return c.flush()
467}
468
469// clientInfo extracts the capability letters the client appends to its -e
470// option, as in "-e.LsfxCIvu".
471func clientInfo(shellCmd string) string {
472 if i := strings.IndexByte(shellCmd, '.'); i >= 0 {
473 return shellCmd[i+1:]
474 }
475 return shellCmd
476}
477
478func checksumByChoice(name string, protocol int) (csumType, error) {
479 if name == "auto" {
480 if protocol >= 30 {
481 return csumMD5, nil
482 }
483 return csumMD4Old, nil
484 }
485 return parseChecksumName(name)
486}
487
488func nameList[T any](items []struct {
489 name string
490 typ T
491}) string {
492 var names []string
493 seen := map[any]bool{}
494 for _, it := range items {
495 if seen[it.typ] {
496 continue
497 }
498 seen[it.typ] = true
499 names = append(names, it.name)
500 }
501 return strings.Join(names, " ")
502}
503
504// negotiateName picks our most preferred name that the client also listed.
505func negotiateName[T any](items []struct {
506 name string
507 typ T
508}, list string) (string, bool) {
509 theirs := strings.Fields(list)
510 for _, it := range items {
511 if slices.Contains(theirs, it.name) {
512 return it.name, true
513 }
514 }
515 return "", false
516}
517
518// handleMsg processes an out-of-band message from the client.
519func (s *session) handleMsg(tag byte, data []byte) error {
520 switch tag {
521 case msgIOError:
522 if len(data) != 4 {
523 return fmt.Errorf("invalid multi-message %d:%d", tag, len(data))
524 }
525 s.addIOError(int32(binary.LittleEndian.Uint32(data)) & ioErrMask)
526 case msgNoop:
527 case msgErrorExit:
528 code := 0
529 if len(data) == 4 {
530 code = int(int32(binary.LittleEndian.Uint32(data)))
531 }
532 return &remoteExitError{code: code}
533 case msgInfo, msgError, msgErrorXfer, msgWarning:
534 s.logger.Info("rsync client message", "msg", strings.TrimSpace(string(data)))
535 case msgNoSend:
536 // The sender couldn't open a file we asked for and has reported
537 // why; nothing will arrive for it.
538 if len(data) != 4 {
539 return fmt.Errorf("invalid multi-message %d:%d", tag, len(data))
540 }
541 case msgSuccess, msgDeleted:
542 default:
543 return fmt.Errorf("unexpected tag %d", tag)
544 }
545 return nil
546}
547
548func (s *session) addIOError(v int32) {
549 s.mu.Lock()
550 s.ioError |= v
551 s.mu.Unlock()
552}
553
554func (s *session) who() string {
555 if s.opts.sender {
556 return "sender"
557 }
558 return "receiver"
559}
560
561func (s *session) send(tag byte, msg string) {
562 if s.c.protocol < 30 {
563 switch tag {
564 case msgError:
565 tag = msgErrorXfer
566 case msgWarning:
567 tag = msgInfo
568 }
569 }
570 if err := s.c.writeMsg(tag, []byte(msg)); err != nil {
571 s.logger.Debug("rsync message not delivered", "err", err)
572 }
573}
574
575func (s *session) info(format string, args ...any) {
576 s.send(msgInfo, fmt.Sprintf(format, args...)+"\n")
577}
578
579func (s *session) warning(format string, args ...any) {
580 s.send(msgWarning, fmt.Sprintf(format, args...)+"\n")
581}
582
583// xferError reports a failure affecting one file; the run continues and
584// ends with a partial-transfer exit code.
585func (s *session) xferError(format string, args ...any) {
586 s.mu.Lock()
587 s.gotXferError = true
588 s.mu.Unlock()
589 s.send(msgErrorXfer, fmt.Sprintf("rsync: [%s] %s\n", s.who(), fmt.Sprintf(format, args...)))
590}
591
592// fatal reports err to the client and ends the run.
593func (s *session) fatal(code int, err error) error {
594 var remote *remoteExitError
595 if errors.As(err, &remote) {
596 return &ExitError{Code: remote.code, Err: err}
597 }
598 s.logger.Error("rsync session failed", "err", err)
599 if !errors.Is(err, io.EOF) && !errors.Is(err, io.ErrUnexpectedEOF) {
600 s.send(msgError, fmt.Sprintf("rsync: [%s] %v\n", s.who(), err))
601 if s.c.protocol >= 31 {
602 _ = s.c.writeMsgInt(msgErrorExit, int32(code))
603 }
604 _ = s.c.flush()
605 }
606 return &ExitError{Code: code, Err: err}
607}
608
609// finish ends a run that completed, reporting a partial transfer if any
610// file failed.
611func (s *session) finish() error {
612 if err := s.c.flush(); err != nil {
613 return err
614 }
615 s.mu.Lock()
616 ioError, xferErr := s.ioError, s.gotXferError
617 s.mu.Unlock()
618 code := 0
619 switch {
620 case ioError&ioErrGeneral != 0 || xferErr:
621 code = exitPartial
622 case ioError&ioErrVanished != 0:
623 code = exitVanished
624 case ioError&ioErrDelLimit != 0:
625 code = exitDelLimit
626 }
627 if code == 0 {
628 return nil
629 }
630 return &ExitError{Code: code, Err: fmt.Errorf("some files were not transferred (code %d)", code)}
631}
632
633func (s *session) newFlistCodec() *flistCodec {
634 return &flistCodec{
635 protocol: s.c.protocol,
636 varintFlags: s.varintFlags,
637 safeFlist: s.safeFlist,
638 preserveUID: s.opts.preserveUID,
639 preserveGID: s.opts.preserveGID,
640 preserveLinks: s.opts.preserveLinks,
641 preserveDevs: s.opts.preserveDevs,
642 preserveSpecs: s.opts.preserveSpecs,
643 preserveAtime: s.opts.preserveAtimes,
644 alwaysSum: s.opts.alwaysChecksum,
645 sumLen: s.cs.file.size(),
646 recurse: s.opts.recurse,
647 relative: s.opts.relative,
648 }
649}
650
651// recvFilterList reads the client's filter rules (exclude.c:recv_filter_list).
652func (s *session) recvFilterList() error {
653 receiverWants := s.opts.pruneEmptyDirs || (s.opts.deleteMode && (!s.opts.deleteExcluded || s.c.protocol >= 29))
654 if s.opts.sender || receiverWants {
655 for {
656 n, err := s.c.readInt32()
657 if err != nil {
658 return err
659 }
660 if n == 0 {
661 break
662 }
663 if n < 0 || n >= maxPathLen*2 {
664 return s.fatal(exitProtocol, fmt.Errorf("filter rule too long: %d", n))
665 }
666 line := make([]byte, n)
667 if err := s.c.readFull(line); err != nil {
668 return err
669 }
670 if err := s.filters.parse(string(line), s.c.protocol < 29); err != nil {
671 return s.fatal(exitSyntax, err)
672 }
673 }
674 }
675 if s.opts.cvsExclude && (s.opts.sender || s.c.protocol < 29) {
676 if err := s.filters.parse(":C", false); err != nil {
677 return s.fatal(exitSyntax, err)
678 }
679 s.filters.rules = append(s.filters.rules, s.filters.cvsIgnoreRules()...)
680 }
681 return nil
682}
683
684type ndxRequest struct {
685 ndx int32
686 iflags int
687 fnamecmpType byte
688 xname string
689 hasXname bool
690}
691
692// readNdxAndAttrs reads the next file index and its item flags
693// (rsync.c:read_ndx_and_attrs).
694func (s *session) readNdxAndAttrs(count int) (ndxRequest, error) {
695 c := s.c
696 for {
697 ndx, err := c.readNdx()
698 if err != nil {
699 return ndxRequest{}, err
700 }
701 if ndx == ndxDone {
702 return ndxRequest{ndx: ndx}, nil
703 }
704 if ndx == ndxDelStats {
705 var counts [5]int32
706 for i := range counts {
707 if counts[i], err = c.readVarint(); err != nil {
708 return ndxRequest{}, err
709 }
710 }
711 if s.opts.sender {
712 // Echo the receiver's deletion stats back to the client.
713 if err := s.writeDelStats(counts); err != nil {
714 return ndxRequest{}, err
715 }
716 }
717 continue
718 }
719 if ndx < 0 || int(ndx) > count {
720 return ndxRequest{}, s.fatal(exitProtocol, fmt.Errorf("invalid file index: %d (%d - %d)", ndx, ndxDone, count-1))
721 }
722 req := ndxRequest{ndx: ndx, iflags: itemTransfer | itemMissingData, fnamecmpType: fnamecmpFname}
723 if c.protocol >= 29 {
724 v, err := c.readShortint()
725 if err != nil {
726 return ndxRequest{}, err
727 }
728 req.iflags = int(v)
729 }
730 // The protocol 29 keep-alive.
731 if c.protocol < 30 && int(ndx) == count && req.iflags == itemIsNew {
732 continue
733 }
734 if int(ndx) == count {
735 return ndxRequest{}, s.fatal(exitProtocol, fmt.Errorf("invalid file index: %d", ndx))
736 }
737 if req.iflags&itemBasisTypeFollows != 0 {
738 b, err := c.readByte()
739 if err != nil {
740 return ndxRequest{}, err
741 }
742 req.fnamecmpType = b
743 }
744 if req.iflags&itemXnameFollows != 0 {
745 if req.xname, err = c.readVstring(maxPathLen); err != nil {
746 return ndxRequest{}, err
747 }
748 req.hasXname = true
749 }
750 return req, nil
751 }
752}
753
754func (s *session) writeNdxAndAttrs(req ndxRequest) error {
755 if err := s.c.writeNdx(req.ndx); err != nil {
756 return err
757 }
758 if s.c.protocol < 29 {
759 return nil
760 }
761 if err := s.c.writeShortint(uint16(req.iflags)); err != nil {
762 return err
763 }
764 if req.iflags&itemBasisTypeFollows != 0 {
765 if err := s.c.writeByte(req.fnamecmpType); err != nil {
766 return err
767 }
768 }
769 if req.iflags&itemXnameFollows != 0 {
770 return s.c.writeVstring(req.xname)
771 }
772 return nil
773}
774
775func (s *session) writeDelStats(counts [5]int32) error {
776 if err := s.c.writeNdx(ndxDelStats); err != nil {
777 return err
778 }
779 for _, v := range counts {
780 if err := s.c.writeVarint(v); err != nil {
781 return err
782 }
783 }
784 return nil
785}
786
787// readFinalGoodbye waits for the client's end-of-run index
788// (main.c:read_final_goodbye).
789func (s *session) readFinalGoodbye(sender bool) error {
790 c := s.c
791 var ndx int32
792 var err error
793 if c.protocol < 29 {
794 ndx, err = c.readInt32()
795 } else {
796 var req ndxRequest
797 req, err = s.readNdxAndAttrs(0)
798 ndx = req.ndx
799 if err == nil && c.protocol >= 31 && ndx == ndxDone && sender {
800 if err = c.writeNdx(ndxDone); err != nil {
801 return err
802 }
803 req, err = s.readNdxAndAttrs(0)
804 ndx = req.ndx
805 }
806 }
807 if err != nil {
808 return err
809 }
810 if ndx != ndxDone {
811 return s.fatal(exitProtocol, fmt.Errorf("invalid packet at end of run (%d)", ndx))
812 }
813 return nil
814}