GODRIVER-3302 Handle malformatted message length properly. #1758

qingyang-hu · 2024-08-14T22:17:39Z

Summary

Add a sanity check for the message size
Update the background read logic

Background & Motivation

connection.read has two places that can raise timeout errors, one is in reading the message header, and the other is in reading the remaining body of the message.
The customer has observed an "incomplete read of full message" before the panic. In this scenario, we should not call connection.read directly in the bgRead to start over the reading because the message header with the message size has already been read. Instead, we ought to head towards the message body with the given size.
On the other hand, we need to start over in the bgRead if the header is not read at all, but cease processing if the header is only partially read to reduce the complexity.

mongodb-drivers-pr-bot · 2024-08-14T22:52:17Z

API Change Report

No changes found!

prestonvasquez · 2024-08-21T22:05:53Z

x/mongo/driver/topology/pool.go

+		}
+		size -= 4
+	}
+	_, err = io.CopyN(ioutil.Discard, conn.nc, int64(size))


ioutil is deprecated, should use io.Discard

Suggested change

_, err = io.CopyN(ioutil.Discard, conn.nc, int64(size))

_, err = io.CopyN(io.Discard, conn.nc, int64(size))

prestonvasquez · 2024-08-21T22:39:51Z

x/mongo/driver/topology/connection.go

 	if err != nil {
+		if l := size - 4 - int32(n); l > 0 && needToWait(err) {


Suggest renaming l to remainingBytes.

prestonvasquez · 2024-08-21T22:40:13Z

x/mongo/driver/topology/connection.go

 	// read before returning the connection to the pool.
-	awaitingResponse bool
+	awaitingResponse *int32


Suggest renaming awaitingResponse to awaitRemainingBytes.

prestonvasquez · 2024-08-21T22:51:51Z

x/mongo/driver/topology/pool.go

+		var sizeBuf [4]byte
+		_, err = io.ReadFull(conn.nc, sizeBuf[:])


Suggest making this logic a function that can be used here and in the connection read method:

func readWMSize(r io.Reader) (int32, error) { const wireMessageSizePrefix = 4 var wmSizeBytes [wireMessageSizePrefix]byte if _, err := io.ReadFull(r, wmSizeBytes[:]); err != nil { return 0, fmt.Errorf("error reading the message size: %w", err) } size := (int32(wmSizeBytes[0])) | (int32(wmSizeBytes[1]) << 8) | (int32(wmSizeBytes[2]) << 16) | (int32(wmSizeBytes[3]) << 24) if size < 4 { return 0, fmt.Errorf("malformed message length: %d", size) } return size, nil }

qingyang-hu · 2024-08-23T21:59:45Z

x/mongo/driver/topology/connection.go

@@ -115,12 +115,6 @@ func newConnection(addr address.Address, opts ...ConnectionOption) *connection {
 	return c
 }

-// DriverConnectionID returns the driver connection ID.


Moved DriverConnectionID down with other public methods.

qingyang-hu · 2024-08-23T22:02:03Z

x/mongo/driver/topology/connection.go

+func (c *connection) cancellationListenerCallback() {
+	_ = c.close()
+}
+
 func transformNetworkError(ctx context.Context, originalError error, contextDeadlineUsed bool) error {


Moved transformNetworkError closer to the caller.

qingyang-hu · 2024-08-23T22:06:54Z

x/mongo/driver/topology/connection.go

@@ -537,10 +589,6 @@ func (c *connection) setCanStream(canStream bool) {
 	c.canStream = canStream
 }

-func (c initConnection) supportsStreaming() bool {


Merged in (initConnection).SupportsStreaming().

qingyang-hu · 2024-08-23T22:10:27Z

x/mongo/driver/topology/connection.go

@@ -833,39 +895,6 @@ func (c *Connection) DriverConnectionID() uint64 {
 	return c.connection.DriverConnectionID()
 }

-func configureTLS(ctx context.Context,


Moved closer to the caller.

qingyang-hu · 2024-08-23T22:11:15Z

x/mongo/driver/topology/connection.go

@@ -919,11 +948,3 @@ func (c *cancellListener) StopListening() bool {
 	c.done <- struct{}{}
 	return c.aborted
 }
-
-func (c *connection) OIDCTokenGenID() uint64 {


Moved closer with other *connection methods.

prestonvasquez

LGTM, great tests!

blink1073 · 2024-08-29T17:46:16Z

drivers-pr-bot please backport to master

matthewdale · 2024-09-12T22:21:41Z

x/mongo/driver/topology/connection.go

@@ -461,36 +506,43 @@ func (c *connection) read(ctx context.Context) (bytesRead []byte, errMsg string,
 		}
 	}()

+	needToWait := func(err error) bool {


Optional: Consider a more descriptive name, like isCSOTTimeout.

matthewdale · 2024-09-12T22:29:25Z

x/mongo/driver/topology/connection.go

+	size := (int32(wmSizeBytes[0])) |
+		(int32(wmSizeBytes[1]) << 8) |
+		(int32(wmSizeBytes[2]) << 16) |
+		(int32(wmSizeBytes[3]) << 24)


Optional: Use the binary package instead.

Suggested change

size := (int32(wmSizeBytes[0])) |

(int32(wmSizeBytes[1]) << 8) |

(int32(wmSizeBytes[2]) << 16) |

(int32(wmSizeBytes[3]) << 24)

size := int32(binary.LittleEndian.Uint32(wmSizeBytes[:]))

matthewdale · 2024-09-12T22:36:19Z

x/mongo/driver/topology/pool_test.go

+			errCh = make(chan error)
+
+			var err error
+			socket, err = net.Listen("unix", sockPath)


Ideally we should use TCP sockets to more accurately test the real-world use case of the connection pool. Is it possible to use TCP instead of Unix sockets?

Consider using the bootstrapConnections helper used elsewhere in pool_test.go, which also uses a connection teardown pattern that doesn't require using sleeps at the end of the connection handler.

matthewdale · 2024-09-13T04:50:19Z

x/mongo/driver/topology/pool_test.go

+		var errCh chan error
+		BGReadCallback = func(addr string, start, read time.Time, errs []error, connClosed bool) {
+			defer close(errCh)
+
+			for _, err := range errs {
+				errCh <- err
+			}
+		}


Sharing this logic between all the subtests will make it easy to introduce unintentional interaction between subtests. We should set the callback for each subtest using test-specific channels. We should also set BGReadCallback back to nil at the end of each subtest to prevent interaction between tests.

E.g.

t.Run("subtest", func(t *testing.T) { var errCh chan error BGReadCallback = func(...) { // ... } t.Cleanup(func() { BGReadCallback = nil }) // Make test assertions. })

matthewdale · 2024-09-13T05:03:25Z

x/mongo/driver/topology/pool.go

+	}
+	_, err = io.CopyN(io.Discard, conn.nc, int64(size))
+	if err != nil {
+		err = fmt.Errorf("error reading message of %d: %w", size, err)


Optional: This error message is a bit confusing. Consider a clearer error message.

Suggested change

err = fmt.Errorf("error reading message of %d: %w", size, err)

err = fmt.Errorf("error discarding %d byte message: %w", size, err)

matthewdale · 2024-09-13T05:13:57Z

x/mongo/driver/topology/pool_test.go

@@ -1122,6 +1126,226 @@ func TestPool(t *testing.T) {
 			p.close(context.Background())
 		})
 	})
+	t.Run("bgRead", func(t *testing.T) {


Optional: Consider moving the bgRead subtests to a separate test function to allow de-indenting them one level.

matthewdale

I recommend refactoring the subtests in TestBackgroundRead to simplify the assertions and speed up the tests. Here is the general recommended pattern:

cleanup := make(chan struct{})
defer close(cleanup)
addr := bootstrapConnections(t, 1, func(nc net.Conn) {
	defer func() {
		<-cleanup
		_ = nc.Close()
	}()

	// Write to the connection here.
})

// Test logic here.

var bgErr error
select {
case bgErr = <-errCh:
case <-time.After(100 * time.Millisecond):
	t.Fatal("did not receive expected error after waiting for 100ms")
}
assert.EqualError(t, bgErr, "<expected error string>")

p.close(context.Background())

matthewdale · 2024-09-13T17:38:56Z

x/mongo/driver/topology/pool_test.go

+		go func(t *testing.T) {
+		}(t)


This empty func can be removed.

Suggested change

go func(t *testing.T) {

}(t)

matthewdale · 2024-09-13T17:49:26Z

x/mongo/driver/topology/pool_test.go

+		wg := &sync.WaitGroup{}
+		wg.Add(1)
+		addr := bootstrapConnections(t, 1, func(nc net.Conn) {
+			t.Helper()


Since this func is only part of this one test, we shouldn't use t.Helper() here because it will obscure the actual error location. That applies for all other anonymous funcs passed to bootstrapConnections in this test.

matthewdale · 2024-09-13T18:47:04Z

x/mongo/driver/topology/pool_test.go

+			var err error
+			_, err = nc.Write([]byte{12, 0, 0, 0, 0, 0, 0, 0, 1})
+			noerr(t, err)
+			time.Sleep(1500 * time.Millisecond)


Since we're using all in-process connections, we should be able to reduce the sleeps and timeouts, which will speed up these tests significantly. I recommend a 50ms sleep here and 10ms timeout in csot.MakeTimeoutContext, which is similar to the timeout values in other pool tests.

The same recommendation applies to all subtests in TestBackgroundRead.

matthewdale · 2024-09-13T18:48:47Z

x/mongo/driver/topology/pool_test.go

+			time.Sleep(1500 * time.Millisecond)
+			_, err = nc.Write([]byte{2, 3, 4})
+			noerr(t, err)
+			time.Sleep(1500 * time.Millisecond)


We can replace the sleep at the end of the connection handler with a "cleanup" signal, which will speed up the tests significantly.

E.g.

cleanup := make(chan struct{}) defer close(cleanup) addr := bootstrapConnections(t, 1, func(nc net.Conn) { defer func() { <-cleanup _ = nc.Close() }() var err error _, err = nc.Write([]byte{12, 0, 0, 0, 0, 0, 0, 0, 1}) // ... })

The same recommendation applies to all subtests in TestBackgroundRead.

matthewdale · 2024-09-13T19:01:58Z

x/mongo/driver/topology/pool_test.go

+		wg := &sync.WaitGroup{}
+		wg.Add(1)


If we use a "cleanup" channel as previously recommended, this WaitGroup will be unnecessary and can be removed.

matthewdale · 2024-09-13T19:02:49Z

x/mongo/driver/topology/pool_test.go

+		wg.Wait()
+		p.close(context.Background())
+		errs := []string{
+			"error discarding 3 byte message: EOF",
+		}
+		for i := 0; true; i++ {
+			err, ok := <-errCh


If we apply the previous recommended changes, we can simplify the assertion logic here.

var bgErr error select { case bgErr = <-errCh: case <-time.After(100 * time.Millisecond): t.Fatal("did not receive expected error after waiting for 100ms") } assert.EqualError(t, bgErr, "error discarding 3 byte message: EOF") p.close(context.Background())

matthewdale · 2024-09-13T19:07:18Z

x/mongo/driver/topology/pool_test.go

+		regex := regexp.MustCompile(
+			`^connection\(.*\[-\d+\]\) incomplete read of full message: context deadline exceeded: read tcp 127.0.0.1:.*->127.0.0.1:.*: i\/o timeout$`,
+		)
+		assert.True(t, regex.MatchString(err.Error()), "mismatched err: %v", err)


Optional: Consider printing the regex in the error message to make troubleshooting these failures easier.

This recommendation applies to all similar assertions in the TestBackgroundRead subtests.

Suggested change

assert.True(t, regex.MatchString(err.Error()), "mismatched err: %v", err)

assert.True(t, regex.MatchString(err.Error()), "error %q does not match pattern %q", err, regex)

matthewdale · 2024-09-16T21:43:44Z

x/mongo/driver/topology/pool_test.go

+					conn, err := net.Dial("tcp", addr.String())
+					noerr(t, err)
+					return newLimitConn(conn, 10), nil


What is the point of using newLimitConn here? Is the test trying to create timeout conditions or connection closed conditions?

You are correct. We no longer need it.

matthewdale · 2024-09-16T21:50:14Z

x/mongo/driver/topology/pool_test.go

+		for i, err := range bgErrs {
+			if i < len(wantErrs) {
+				assert.EqualError(t, err, wantErrs[i], "mismatched err: %v", err)


This assertion loop will pass if len(bgErrs) == 0, even if wantErrs contains expected error messages. We should fix and simplify the assertion logic.

E.g.

require.Len(t, bgErrs, "expected 1 error") assert.EqualError(t, bgErrs[0], "error discarding 3 byte message: EOF")

matthewdale · 2024-09-16T21:54:01Z

x/mongo/driver/topology/pool_test.go

+		for i, err := range bgErrs {
+			if i < len(wantErrs) {
+				assert.True(t, wantErrs[i].MatchString(err.Error()), "error %q does not match pattern %q", err, wantErrs[i])


This assertion loop will pass if len(bgErrs) == 0, even if wantErrs contains expected error messages. We should fix and simplify the assertion logic.

E.g.

require.Len(t, bgErrs, "expected 1 error") wantErrPattern := regexp.MustCompile(`^error discarding 6 byte message: read tcp 127.0.0.1:.*->127.0.0.1:.*: i\/o timeout$`) gotErr := bgErrs[0] assert.True(t, wantErrPattern.MatchString(gotErr), "error %q does not match pattern %q", gotErr, wantErrPattern)

matthewdale · 2024-09-16T21:57:57Z

x/mongo/driver/topology/pool_test.go

+		p := newPool(
+			poolConfig{},
+			WithDialer(func(Dialer) Dialer {
+				return DialerFunc(func(context.Context, string, string) (net.Conn, error) {
+					return net.Dial("tcp", addr.String())
+				})
+			}),
+		)


Optional: Since there's no special dialing logic added here, we can pass addr in the poolConfig instead of overriding the dialer.

Suggested change

p := newPool(

poolConfig{},

WithDialer(func(Dialer) Dialer {

return DialerFunc(func(context.Context, string, string) (net.Conn, error) {

return net.Dial("tcp", addr.String())

})

}),

)

p := newPool(poolConfig{

Address: address.Address(addr.String()),

})

matthewdale · 2024-09-16T21:59:11Z

x/mongo/driver/topology/pool_test.go

+		p := newPool(
+			poolConfig{},
+			WithDialer(func(Dialer) Dialer {
+				return DialerFunc(func(context.Context, string, string) (net.Conn, error) {
+					return net.Dial("tcp", addr.String())
+				})
+			}),
+		)


Optional: Since there's no special dialing logic added here, we can pass addr in the poolConfig instead of overriding the dialer.

Suggested change

p := newPool(

poolConfig{},

WithDialer(func(Dialer) Dialer {

return DialerFunc(func(context.Context, string, string) (net.Conn, error) {

return net.Dial("tcp", addr.String())

})

}),

)

p := newPool(poolConfig{

Address: address.Address(addr.String()),

})

matthewdale · 2024-09-16T22:31:49Z

x/mongo/driver/topology/pool_test.go

+		assert.Nil(t, conn.awaitRemainingBytes, "conn.awaitRemainingBytes should be nil")
+		close(errsCh) // this line causes a double close if BGReadCallback is ever called.
+	})
+	t.Run("timeout on reading the message header", func(t *testing.T) {


This test seems to cover the scenario where the operation times out while waiting for the header, then bgRead times out waiting for the full message (because not enough bytes are written). We should also test the scenario where the operation times out while waiting for the header, but then bgRead succeeds.

E.g.

t.Run("timeout reading message header, successful background read", func(t *testing.T) { // ... addr := bootstrapConnections(t, 1, func(nc net.Conn) { defer func() { <-cleanup _ = nc.Close() }() // Wait until the operation times out, then write the full message // length. time.Sleep(timeout * 2) _, err := nc.Write([]byte{10, 0, 0, 0, 0, 0, 0, 0, 0, 0}) noerr(t, err) }) // ... assert.Len(t, bgErrs, 0, "expected no errors from bgRead") })

To differentiate it from the existing test, I recommend updating the test description and adding a comment:

t.Run("timeout reading message header, incomplete background read", func(t *testing.T) { // ... addr := bootstrapConnections(t, 1, func(nc net.Conn) { defer func() { <-cleanup _ = nc.Close() }() // Wait until the operation times out, then write an incomplete // message. time.Sleep(timeout * 5) _, err := nc.Write([]byte{10, 0, 0, 0, 0, 0, 0, 0}) noerr(t, err) }) // ... })

I will also add cases for "timeout reading message header, incomplete head during background read" and "timeout reading full message, successful background read".

matthewdale

Looks good! 👍

mongodb-drivers-pr-bot · 2024-09-17T18:10:25Z

Sorry, unable to cherry-pick to master, please backport manually. Here are approximate instructions:

Checkout backport branch and update it.

git checkout -b cherry-pick-master-be25b9a26aff54fe27210f86d00db5d573452643 master

git fetch origin be25b9a26aff54fe27210f86d00db5d573452643

Cherry pick the first parent branch of the this PR on top of the older branch:

git cherry-pick -x -m1 be25b9a26aff54fe27210f86d00db5d573452643

You will likely have some merge/cherry-pick conflicts here, fix them and commit:

git commit -am {message}

Push to a named branch:

git push origin cherry-pick-master-be25b9a26aff54fe27210f86d00db5d573452643

Create a PR against branch master. I would have named this PR:

"GODRIVER-3302 Handle malformatted message length properly. (#1758) [master]"

(cherry picked from commit be25b9a)

…aster] (#1817)

GODRIVER-3302 Handle malformatted message length properly.

93d8ee1

mongodb-drivers-pr-bot bot added the priority-3-low Low Priority PR for Review label Aug 14, 2024

qingyang-hu requested a review from prestonvasquez August 15, 2024 13:31

qingyang-hu marked this pull request as ready for review August 15, 2024 13:33

qingyang-hu marked this pull request as draft August 15, 2024 17:41

fix spelling

be66ac6

qingyang-hu force-pushed the godriver3302 branch 5 times, most recently from 03121d0 to 7a55a9d Compare August 20, 2024 17:09

update the connection pool background read logic

ddbd3e9

qingyang-hu force-pushed the godriver3302 branch from 7a55a9d to ddbd3e9 Compare August 20, 2024 19:11

qingyang-hu marked this pull request as ready for review August 20, 2024 19:54

qingyang-hu requested a review from matthewdale August 20, 2024 19:54

prestonvasquez requested changes Aug 21, 2024

View reviewed changes

code cleanup

b565a23

qingyang-hu commented Aug 23, 2024

View reviewed changes

qingyang-hu requested a review from prestonvasquez August 23, 2024 22:21

prestonvasquez previously approved these changes Aug 27, 2024

View reviewed changes

blink1073 added the priority-2-medium Medium Priority PR for Review label Aug 29, 2024

qingyang-hu removed the priority-3-low Low Priority PR for Review label Aug 30, 2024

blink1073 added priority-1-high High Priority PR for Review and removed priority-2-medium Medium Priority PR for Review labels Sep 12, 2024

matthewdale requested changes Sep 13, 2024

View reviewed changes

improvements

2d988ed

qingyang-hu dismissed prestonvasquez’s stale review via 2d988ed September 13, 2024 15:10

qingyang-hu requested a review from matthewdale September 13, 2024 15:53

matthewdale requested changes Sep 13, 2024

View reviewed changes

improve test logic

9bea0eb

qingyang-hu requested a review from matthewdale September 13, 2024 22:32

Merge branch 'v1' into godriver3302

03aa027

matthewdale requested changes Sep 16, 2024

View reviewed changes

improve tests

39f3021

qingyang-hu requested a review from matthewdale September 17, 2024 16:02

matthewdale approved these changes Sep 17, 2024

View reviewed changes

qingyang-hu merged commit be25b9a into mongodb:v1 Sep 17, 2024
30 of 33 checks passed

qingyang-hu added a commit that referenced this pull request Sep 17, 2024

GODRIVER-3302 Handle malformatted message length properly. (#1758)

e52787c

(cherry picked from commit be25b9a)

qingyang-hu added a commit that referenced this pull request Sep 17, 2024

GODRIVER-3302 Handle malformatted message length properly. (#1758)

f5c3563

qingyang-hu added a commit that referenced this pull request Sep 17, 2024

GODRIVER-3302 Handle malformatted message length properly. (#1758) [m…

e556841

…aster] (#1817)

This pull request was closed.

	_, err = io.CopyN(ioutil.Discard, conn.nc, int64(size))
	_, err = io.CopyN(io.Discard, conn.nc, int64(size))

		if err != nil {
		if l := size - 4 - int32(n); l > 0 && needToWait(err) {

		var sizeBuf [4]byte
		_, err = io.ReadFull(conn.nc, sizeBuf[:])

	err = fmt.Errorf("error reading message of %d: %w", size, err)
	err = fmt.Errorf("error discarding %d byte message: %w", size, err)

	assert.True(t, regex.MatchString(err.Error()), "mismatched err: %v", err)
	assert.True(t, regex.MatchString(err.Error()), "error %q does not match pattern %q", err, regex)

GODRIVER-3302 Handle malformatted message length properly. #1758

GODRIVER-3302 Handle malformatted message length properly. #1758

Conversation

qingyang-hu commented Aug 14, 2024 • edited Loading

Summary

Background & Motivation

mongodb-drivers-pr-bot bot commented Aug 14, 2024 • edited Loading

API Change Report

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

qingyang-hu Aug 23, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

qingyang-hu Aug 23, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

prestonvasquez left a comment • edited Loading

Choose a reason for hiding this comment

blink1073 commented Aug 29, 2024

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

matthewdale left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

matthewdale Sep 16, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

matthewdale left a comment

Choose a reason for hiding this comment

mongodb-drivers-pr-bot bot commented Sep 17, 2024

qingyang-hu commented Aug 14, 2024 •

edited

Loading

mongodb-drivers-pr-bot bot commented Aug 14, 2024 •

edited

Loading

qingyang-hu Aug 23, 2024 •

edited

Loading

qingyang-hu Aug 23, 2024 •

edited

Loading

prestonvasquez left a comment •

edited

Loading

matthewdale Sep 16, 2024 •

edited

Loading