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