fix(rsync): ensure no leaked routines on failed upload

Commit `1ac67bc`

Antonio Mika · 2026-10-09 10:56:11 -0400 EDT

parent a1973f3

fix(rsync): ensure no leaked routines on failed upload

3 files changed, +49, -12

M pkg/rsync/conn.go

M pkg/rsync/receiver.go

M pkg/rsync/wire_test.go

M pkg/rsync/conn.go

+9, -5

......@@ -32,7 +32,6 @@ type muxWriter struct {
	multiplex bool
	written   int64
	err       error
	closed    bool
}

type muxItem struct {

......@@ -115,11 +114,16 @@ func (m *muxWriter) Flush() error {
	return m.err
}

// Close flushes and stops the writer goroutine.
// errMuxClosed is returned to writers that outlive Close.
var errMuxClosed = errors.New("rsync: write after close")

// Close flushes and stops the writer goroutine. Later writes fail.
func (m *muxWriter) Close() error {
	err := m.Flush()
	m.mu.Lock()
	m.closed = true
	if m.err == nil {
		m.err = errMuxClosed
	}
	m.cond.Broadcast()
	m.mu.Unlock()
	return err

......@@ -139,10 +143,10 @@ func (m *muxWriter) pump() {
	m.mu.Lock()
	defer m.mu.Unlock()
	for {
		for !m.closed && m.err == nil && (!m.kick || len(m.queue) == 0) {
		for m.err == nil && (!m.kick || len(m.queue) == 0) {
			m.cond.Wait()
		}
		if m.closed || m.err != nil {
		if m.err != nil {
			m.queue = nil
			m.cond.Broadcast()
			return

M pkg/rsync/receiver.go

+3, -0

......@@ -205,6 +205,9 @@ func (g *generator) run() error {
		if f.inactive {
			continue
		}
		if err := g.ctx.Err(); err != nil {
			return err
		}
		if err := g.recvGenerator(int32(i), f, false); err != nil {
			return err
		}

M pkg/rsync/wire_test.go

+37, -7

......@@ -5,23 +5,25 @@ import (
	"io"
	"slices"
	"testing"
	"time"
)

type loopback struct {
	bytes.Buffer
}

func newTestConn(protocol int) (*conn, *loopback) {
func newTestConn(t *testing.T, protocol int) (*conn, *loopback) {
	var buf loopback
	c := newConn(&buf)
	c.protocol = protocol
	t.Cleanup(func() { _ = c.close() })
	return c, &buf
}

func TestVarintRoundTrip(t *testing.T) {
	values := []int32{0, 1, 0x7f, 0x80, 0xff, 0x3fff, 0x4000, 0x1fffff, 0x200000, 0x0fffffff, 0x10000000, 0x7fffffff, -1, -2, -0x80000000}
	for _, v := range values {
		c, _ := newTestConn(31)
		c, _ := newTestConn(t, 31)
		if err := c.writeVarint(v); err != nil {
			t.Fatal(err)
		}

......@@ -42,7 +44,7 @@ func TestVarlongRoundTrip(t *testing.T) {
	values := []int64{0, 1, 0xffffff, 0x1000000, 0x7fffffff, 0x80000000, 1 << 40, 1<<62 + 12345, -1}
	for _, min := range []int{3, 4} {
		for _, v := range values {
			c, _ := newTestConn(31)
			c, _ := newTestConn(t, 31)
			if err := c.writeVarlong(v, min); err != nil {
				t.Fatal(err)
			}

......@@ -59,7 +61,7 @@ func TestVarlongRoundTrip(t *testing.T) {

func TestLongintRoundTrip(t *testing.T) {
	for _, v := range []int64{0, 0x7fffffff, 0x80000000, 1 << 40, -5} {
		c, _ := newTestConn(29)
		c, _ := newTestConn(t, 29)
		if err := c.writeLongint(v); err != nil {
			t.Fatal(err)
		}

......@@ -76,7 +78,7 @@ func TestLongintRoundTrip(t *testing.T) {
func TestNdxRoundTrip(t *testing.T) {
	seq := []int32{0, 1, 2, 5, 300, 299, 40000, 70000, 3, ndxDone, ndxDelStats, ndxFlistEOF, 7, ndxDone}
	for _, proto := range []int{29, 30} {
		c, _ := newTestConn(proto)
		c, _ := newTestConn(t, proto)
		for _, v := range seq {
			if err := c.writeNdx(v); err != nil {
				t.Fatal(err)

......@@ -95,7 +97,7 @@ func TestNdxRoundTrip(t *testing.T) {
}

func TestMultiplexedMessages(t *testing.T) {
	c, buf := newTestConn(31)
	c, buf := newTestConn(t, 31)
	if err := c.w.setMultiplex(true); err != nil {
		t.Fatal(err)
	}

......@@ -106,7 +108,7 @@ func TestMultiplexedMessages(t *testing.T) {
		t.Fatal(err)
	}

	r, _ := newTestConn(31)
	r, _ := newTestConn(t, 31)
	r.r.r.Reset(bytes.NewReader(buf.Bytes()))
	r.r.multiplex = true
	var msgs []string

......@@ -130,6 +132,34 @@ func TestMultiplexedMessages(t *testing.T) {
	}
}

// A generator can still be writing when the session gives up and closes the
// writer; it has to get an error instead of waiting on a queue nobody drains.
func TestMuxWriterAfterClose(t *testing.T) {
	m := newMuxWriter(io.Discard)
	if err := m.Close(); err != nil {
		t.Fatal(err)
	}
	done := make(chan error, 1)
	go func() {
		chunk := make([]byte, muxBatchSize)
		for range 2 * muxMaxQueued / muxBatchSize {
			if _, err := m.Write(chunk); err != nil {
				done <- err
				return
			}
		}
		done <- m.Flush()
	}()
	select {
	case err := <-done:
		if err == nil {
			t.Fatal("writes after Close succeeded")
		}
	case <-time.After(5 * time.Second):
		t.Fatal("write after Close blocked")
	}
}

func TestChecksum1(t *testing.T) {
	// Bytes above 0x7f count as negative, as in rsync's C code.
	if got := checksum1([]byte{0xff, 1, 2, 3, 4, 5}); got != 0x001d000e {

...	...		@@ -205,6 +205,9 @@ func (g *generator) run() error {
205	205		if f.inactive {
206	206		continue
207	207		}
	208	+	if err := g.ctx.Err(); err != nil {
	209	+	return err
	210	+	}
208	211		if err := g.recvGenerator(int32(i), f, false); err != nil {
209	212		return err
210	213		}

...	...		@@ -130,6 +132,34 @@ func TestMultiplexedMessages(t *testing.T) {
130	132		}
131	133		}
132	134
	135	+	// A generator can still be writing when the session gives up and closes the
	136	+	// writer; it has to get an error instead of waiting on a queue nobody drains.
	137	+	func TestMuxWriterAfterClose(t *testing.T) {
	138	+	m := newMuxWriter(io.Discard)
	139	+	if err := m.Close(); err != nil {
	140	+	t.Fatal(err)
	141	+	}
	142	+	done := make(chan error, 1)
	143	+	go func() {
	144	+	chunk := make([]byte, muxBatchSize)
	145	+	for range 2 * muxMaxQueued / muxBatchSize {
	146	+	if _, err := m.Write(chunk); err != nil {
	147	+	done <- err
	148	+	return
	149	+	}
	150	+	}
	151	+	done <- m.Flush()
	152	+	}()
	153	+	select {
	154	+	case err := <-done:
	155	+	if err == nil {
	156	+	t.Fatal("writes after Close succeeded")
	157	+	}
	158	+	case <-time.After(5 * time.Second):
	159	+	t.Fatal("write after Close blocked")
	160	+	}
	161	+	}
	162	+
133	163		func TestChecksum1(t *testing.T) {
134	164		// Bytes above 0x7f count as negative, as in rsync's C code.
135	165		if got := checksum1([]byte{0xff, 1, 2, 3, 4, 5}); got != 0x001d000e {

...	...		@@ -32,7 +32,6 @@ type muxWriter struct {
32	32		multiplex bool
33	33		written int64
34	34		err error
35		-	closed bool
36	35		}
37	36
38	37		type muxItem struct {

...	...		@@ -115,11 +114,16 @@ func (m *muxWriter) Flush() error {
115	114		return m.err
116	115		}
117	116
118		-	// Close flushes and stops the writer goroutine.
	117	+	// errMuxClosed is returned to writers that outlive Close.
	118	+	var errMuxClosed = errors.New("rsync: write after close")
	119	+
	120	+	// Close flushes and stops the writer goroutine. Later writes fail.
119	121		func (m *muxWriter) Close() error {
120	122		err := m.Flush()
121	123		m.mu.Lock()
122		-	m.closed = true
	124	+	if m.err == nil {
	125	+	m.err = errMuxClosed
	126	+	}
123	127		m.cond.Broadcast()
124	128		m.mu.Unlock()
125	129		return err

...	...		@@ -139,10 +143,10 @@ func (m *muxWriter) pump() {
139	143		m.mu.Lock()
140	144		defer m.mu.Unlock()
141	145		for {
142		-	for !m.closed && m.err == nil && (!m.kick \|\| len(m.queue) == 0) {
	146	+	for m.err == nil && (!m.kick \|\| len(m.queue) == 0) {
143	147		m.cond.Wait()
144	148		}
145		-	if m.closed \|\| m.err != nil {
	149	+	if m.err != nil {
146	150		m.queue = nil
147	151		m.cond.Broadcast()
148	152		return

...	...		@@ -5,23 +5,25 @@ import (
5	5		"io"
6	6		"slices"
7	7		"testing"
	8	+	"time"
8	9		)
9	10
10	11		type loopback struct {
11	12		bytes.Buffer
12	13		}
13	14
14		-	func newTestConn(protocol int) (conn, loopback) {
	15	+	func newTestConn(t testing.T, protocol int) (conn, *loopback) {
15	16		var buf loopback
16	17		c := newConn(&buf)
17	18		c.protocol = protocol
	19	+	t.Cleanup(func() { _ = c.close() })
18	20		return c, &buf
19	21		}
20	22
21	23		func TestVarintRoundTrip(t *testing.T) {
22	24		values := []int32{0, 1, 0x7f, 0x80, 0xff, 0x3fff, 0x4000, 0x1fffff, 0x200000, 0x0fffffff, 0x10000000, 0x7fffffff, -1, -2, -0x80000000}
23	25		for _, v := range values {
24		-	c, _ := newTestConn(31)
	26	+	c, _ := newTestConn(t, 31)
25	27		if err := c.writeVarint(v); err != nil {
26	28		t.Fatal(err)
27	29		}

...	...		@@ -42,7 +44,7 @@ func TestVarlongRoundTrip(t *testing.T) {
42	44		values := []int64{0, 1, 0xffffff, 0x1000000, 0x7fffffff, 0x80000000, 1 << 40, 1<<62 + 12345, -1}
43	45		for _, min := range []int{3, 4} {
44	46		for _, v := range values {
45		-	c, _ := newTestConn(31)
	47	+	c, _ := newTestConn(t, 31)
46	48		if err := c.writeVarlong(v, min); err != nil {
47	49		t.Fatal(err)
48	50		}

...	...		@@ -59,7 +61,7 @@ func TestVarlongRoundTrip(t *testing.T) {
59	61
60	62		func TestLongintRoundTrip(t *testing.T) {
61	63		for _, v := range []int64{0, 0x7fffffff, 0x80000000, 1 << 40, -5} {
62		-	c, _ := newTestConn(29)
	64	+	c, _ := newTestConn(t, 29)
63	65		if err := c.writeLongint(v); err != nil {
64	66		t.Fatal(err)
65	67		}

...	...		@@ -76,7 +78,7 @@ func TestLongintRoundTrip(t *testing.T) {
76	78		func TestNdxRoundTrip(t *testing.T) {
77	79		seq := []int32{0, 1, 2, 5, 300, 299, 40000, 70000, 3, ndxDone, ndxDelStats, ndxFlistEOF, 7, ndxDone}
78	80		for _, proto := range []int{29, 30} {
79		-	c, _ := newTestConn(proto)
	81	+	c, _ := newTestConn(t, proto)
80	82		for _, v := range seq {
81	83		if err := c.writeNdx(v); err != nil {
82	84		t.Fatal(err)

...	...		@@ -95,7 +97,7 @@ func TestNdxRoundTrip(t *testing.T) {
95	97		}
96	98
97	99		func TestMultiplexedMessages(t *testing.T) {
98		-	c, buf := newTestConn(31)
	100	+	c, buf := newTestConn(t, 31)
99	101		if err := c.w.setMultiplex(true); err != nil {
100	102		t.Fatal(err)
101	103		}

...	...		@@ -106,7 +108,7 @@ func TestMultiplexedMessages(t *testing.T) {
106	108		t.Fatal(err)
107	109		}
108	110
109		-	r, _ := newTestConn(31)
	111	+	r, _ := newTestConn(t, 31)
110	112		r.r.r.Reset(bytes.NewReader(buf.Bytes()))
111	113		r.r.multiplex = true
112	114		var msgs []string

Commit 1ac67bc

Commit `1ac67bc`