Make DeadlineConn http.Listener compatible (#20635)

HTTP likes to slap an infinite read deadline on a connection and 
do a blocking read while the response is being written.

This effectively means that a reading deadline becomes the 
request-response deadline.

Instead of enforcing our timeout, we pass it through and keep 
"infinite deadline" is sticky on connections.

However, we still "record" when reads are aborted, so we never overwrite that.

The HTTP server should have `ReadTimeout` and `IdleTimeout` set for the deadline to be effective.

Use --idle-timeout for incoming connections.

internal/deadlineconn/deadlineconn.go

@@ -37,13 +37,23 @@ type DeadlineConn struct {
 	writeDeadline           time.Duration // sets the write deadline on a connection.
 	writeSetAt              time.Time
 	abortReads, abortWrites atomic.Bool // A deadline was set to indicate caller wanted the conn to time out.
+	infReads, infWrites     atomic.Bool
 	mu                      sync.Mutex
 }
 
+// Unwrap will unwrap the connection and remove the deadline if applied.
+// If not a *DeadlineConn, the unmodified net.Conn is returned.
+func Unwrap(c net.Conn) net.Conn {
+	if dc, ok := c.(*DeadlineConn); ok {
+		return dc.Conn
+	}
+	return c
+}
+
 // Sets read deadline
 func (c *DeadlineConn) setReadDeadline() {
 	// Do not set a Read deadline, if upstream wants to cancel all reads.
-	if c.readDeadline <= 0 || c.abortReads.Load() {
+	if c.readDeadline <= 0 || c.abortReads.Load() || c.infReads.Load() {
 		return
 	}
 
@@ -62,7 +72,7 @@ func (c *DeadlineConn) setReadDeadline() {
 
 func (c *DeadlineConn) setWriteDeadline() {
 	// Do not set a Write deadline, if upstream wants to cancel all reads.
-	if c.writeDeadline <= 0 || c.abortWrites.Load() {
+	if c.writeDeadline <= 0 || c.abortWrites.Load() || c.infWrites.Load() {
 		return
 	}
 
@@ -103,31 +113,13 @@ func (c *DeadlineConn) Write(b []byte) (n int, err error) {
 func (c *DeadlineConn) SetDeadline(t time.Time) error {
 	c.mu.Lock()
 	defer c.mu.Unlock()
-	if t.IsZero() {
-		var err error
-		if c.readDeadline == 0 {
-			err = c.Conn.SetReadDeadline(t)
-		}
-		if c.writeDeadline == 0 {
-			if wErr := c.Conn.SetWriteDeadline(t); wErr != nil {
-				return wErr
-			}
-		}
-		c.abortReads.Store(false)
-		c.abortWrites.Store(false)
-		return err
-	}
-	// If upstream sets a deadline in the past, assume it wants to abort reads/writes.
-	if time.Until(t) < 0 {
-		c.abortReads.Store(true)
-		c.abortWrites.Store(true)
-		return c.Conn.SetDeadline(t)
-	}
 
-	c.abortReads.Store(false)
-	c.abortWrites.Store(false)
 	c.readSetAt = time.Now()
 	c.writeSetAt = time.Now()
+	c.abortReads.Store(!t.IsZero() && time.Until(t) < 0)
+	c.abortWrites.Store(!t.IsZero() && time.Until(t) < 0)
+	c.infReads.Store(t.IsZero())
+	c.infWrites.Store(t.IsZero())
 	return c.Conn.SetDeadline(t)
 }
 
@@ -135,15 +127,10 @@ func (c *DeadlineConn) SetDeadline(t time.Time) error {
 // and any currently-blocked Read call.
 // A zero value for t means Read will not time out.
 func (c *DeadlineConn) SetReadDeadline(t time.Time) error {
-	if t.IsZero() && c.readDeadline != 0 {
-		c.abortReads.Store(false)
-		// Keep the deadline we want.
-		return nil
-	}
-
 	c.mu.Lock()
 	defer c.mu.Unlock()
-	c.abortReads.Store(time.Until(t) < 0)
+	c.abortReads.Store(!t.IsZero() && time.Until(t) < 0)
+	c.infReads.Store(t.IsZero())
 	c.readSetAt = time.Now()
 	return c.Conn.SetReadDeadline(t)
 }
@@ -154,14 +141,10 @@ func (c *DeadlineConn) SetReadDeadline(t time.Time) error {
 // some of the data was successfully written.
 // A zero value for t means Write will not time out.
 func (c *DeadlineConn) SetWriteDeadline(t time.Time) error {
-	if t.IsZero() && c.writeDeadline != 0 {
-		c.abortWrites.Store(false)
-		// Keep the deadline we want.
-		return nil
-	}
 	c.mu.Lock()
 	defer c.mu.Unlock()
-	c.abortWrites.Store(time.Until(t) < 0)
+	c.abortWrites.Store(!t.IsZero() && time.Until(t) < 0)
+	c.infWrites.Store(t.IsZero())
 	c.writeSetAt = time.Now()
 	return c.Conn.SetWriteDeadline(t)
 }