(POC) GODRIVER-3414 Complete pending reads on conn checkout

event/monitoring.go

@@ -91,17 +91,20 @@ const (
 
 // strings for pool command monitoring types
 const (
-	PoolCreated        = "ConnectionPoolCreated"
-	PoolReady          = "ConnectionPoolReady"
-	PoolCleared        = "ConnectionPoolCleared"
-	PoolClosedEvent    = "ConnectionPoolClosed"
-	ConnectionCreated  = "ConnectionCreated"
-	ConnectionReady    = "ConnectionReady"
-	ConnectionClosed   = "ConnectionClosed"
-	GetStarted         = "ConnectionCheckOutStarted"
-	GetFailed          = "ConnectionCheckOutFailed"
-	GetSucceeded       = "ConnectionCheckedOut"
-	ConnectionReturned = "ConnectionCheckedIn"
+	PoolCreated                    = "ConnectionPoolCreated"
+	PoolReady                      = "ConnectionPoolReady"
+	PoolCleared                    = "ConnectionPoolCleared"
+	PoolClosedEvent                = "ConnectionPoolClosed"
+	ConnectionCreated              = "ConnectionCreated"
+	ConnectionReady                = "ConnectionReady"
+	ConnectionClosed               = "ConnectionClosed"
+	GetStarted                     = "ConnectionCheckOutStarted"
+	GetFailed                      = "ConnectionCheckOutFailed"
+	GetSucceeded                   = "ConnectionCheckedOut"
+	ConnectionReturned             = "ConnectionCheckedIn"
+	ConnectionPendingReadStarted   = "ConnectionPendingReadStarted"
+	ConnectionPendingReadSucceeded = "ConnectionPendingReadSucceeded"
+	ConnectionPendingReadFailed    = "ConnectionPendingReadFailed"
 )
 
 // MonitorPoolOptions contains pool options as formatted in pool events
@@ -121,9 +124,11 @@ type PoolEvent struct {
 	Reason       string              `json:"reason"`
 	// ServiceID is only set if the Type is PoolCleared and the server is deployed behind a load balancer. This field
 	// can be used to distinguish between individual servers in a load balanced deployment.
-	ServiceID    *primitive.ObjectID `json:"serviceId"`
-	Interruption bool                `json:"interruptInUseConnections"`
-	Error        error               `json:"error"`
+	ServiceID     *primitive.ObjectID `json:"serviceId"`
+	Interruption  bool                `json:"interruptInUseConnections"`
+	Error         error               `json:"error"`
+	RequestID     int32               `json:"requestId"`
+	RemainingTime time.Duration       `json:"remainingTime"`
 }
 
 // PoolMonitor is a function that allows the user to gain access to events occurring in the pool

internal/driverutil/context.go

@@ -0,0 +1,34 @@
+// Copyright (C) MongoDB, Inc. 2025-present.
+//
+// Licensed under the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License. You may obtain
+// a copy of the License at http://www.apache.org/licenses/LICENSE-2.0
+
+package driverutil
+
+import "context"
+
+type ContextKey string
+
+const (
+	ContextKeyHasMaxTimeMS ContextKey = "hasMaxTimeMS"
+	ContextKeyRequestID    ContextKey = "requestID"
+)
+
+func WithValueHasMaxTimeMS(parentCtx context.Context, val bool) context.Context {
+	return context.WithValue(parentCtx, ContextKeyHasMaxTimeMS, val)
+}
+
+func WithRequestID(parentCtx context.Context, requestID int32) context.Context {
+	return context.WithValue(parentCtx, ContextKeyRequestID, requestID)
+}
+
+func HasMaxTimeMS(ctx context.Context) bool {
+	return ctx.Value(ContextKeyHasMaxTimeMS) != nil
+}
+
+func GetRequestID(ctx context.Context) (int32, bool) {
+	val, ok := ctx.Value(ContextKeyRequestID).(int32)
+
+	return val, ok
+}

internal/logger/component.go

@@ -28,6 +28,9 @@ const (
 	ConnectionCheckoutFailed         = "Connection checkout failed"
 	ConnectionCheckedOut             = "Connection checked out"
 	ConnectionCheckedIn              = "Connection checked in"
+	ConnectionPendingReadStarted     = "Pending read started"
+	ConnectionPendingReadSucceeded   = "Pending read succeeded"
+	ConnectionPendingReadFailed      = "Pending read failed"
 	ServerSelectionFailed            = "Server selection failed"
 	ServerSelectionStarted           = "Server selection started"
 	ServerSelectionSucceeded         = "Server selection succeeded"

internal/ptrutil/ptr.go

@@ -0,0 +1,12 @@
+// Copyright (C) MongoDB, Inc. 2024-present.
+//
+// Licensed under the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License. You may obtain
+// a copy of the License at http://www.apache.org/licenses/LICENSE-2.0
+
+package ptrutil
+
+// Ptr will return the memory location of the given value.
+func Ptr[T any](val T) *T {
+	return &val
+}

mongo/integration/csot_test.go

@@ -77,7 +77,7 @@ func TestCSOT_maxTimeMS(t *testing.T) {
 			},
 			sendsMaxTimeMSWithTimeoutMS:       true,
 			sendsMaxTimeMSWithContextDeadline: false,
-			preventsConnClosureWithTimeoutMS:  true,
+			preventsConnClosureWithTimeoutMS:  false,
 		},
 		{
 			desc:        "FindOneAndDelete",
@@ -206,7 +206,7 @@ func TestCSOT_maxTimeMS(t *testing.T) {
 			},
 			sendsMaxTimeMSWithTimeoutMS:       true,
 			sendsMaxTimeMSWithContextDeadline: false,
-			preventsConnClosureWithTimeoutMS:  true,
+			preventsConnClosureWithTimeoutMS:  false,
 		},
 		{
 			desc:        "Watch",
@@ -220,7 +220,7 @@ func TestCSOT_maxTimeMS(t *testing.T) {
 			},
 			sendsMaxTimeMSWithTimeoutMS:       true,
 			sendsMaxTimeMSWithContextDeadline: true,
-			preventsConnClosureWithTimeoutMS:  true,
+			preventsConnClosureWithTimeoutMS:  false,
 			// Change Streams aren't supported on standalone topologies.
 			topologies: []mtest.TopologyKind{
 				mtest.ReplicaSet,