allow retry on some netErrors

Since the commit "Don't return ErrBadConn on a network error"
net.OpError do not return driver.ErrBadConn anymore.

This caused that in some situations the sql package does not retry an
operation when it should.
E.g. when the postgresql server is restarted, a broken pipe error might
happen for the query is done after the server finished the startup
(lib#870).

With this commit driver.ErrBadConn is returned for netErrors when it's
ensured that the server did not already executed the operation.

This is the case when e.g. a netError occur for the call that tries to
send the message to initiate the query to the server.

Author: Fabian Holler

conn.go

@@ -600,7 +600,9 @@ func (cn *conn) gname() string {
 func (cn *conn) simpleExec(q string) (res driver.Result, commandTag string, err error) {
 	b := cn.writeBuf('Q')
 	b.string(q)
-	cn.send(b)
+	if err := cn.send(b); err != nil {
+		return res, commandTag, cn.errHandleFatalNetErrors(err)
+	}
 
 	for {
 		t, r := cn.recv1()
@@ -632,7 +634,10 @@ func (cn *conn) simpleQuery(q string) (res *rows, err error) {
 
 	b := cn.writeBuf('Q')
 	b.string(q)
-	cn.send(b)
+
+	if err := cn.send(b); err != nil {
+		return nil, cn.errHandleFatalNetErrors(err)
+	}
 
 	for {
 		t, r := cn.recv1()
@@ -752,7 +757,7 @@ func decideColumnFormats(colTyps []fieldDesc, forceText bool) (colFmts []format,
 	}
 }
 
-func (cn *conn) prepareTo(q, stmtName string) *stmt {
+func (cn *conn) prepareTo(q, stmtName string) (*stmt, error) {
 	st := &stmt{cn: cn, name: stmtName}
 
 	b := cn.writeBuf('P')
@@ -765,13 +770,16 @@ func (cn *conn) prepareTo(q, stmtName string) *stmt {
 	b.string(st.name)
 
 	b.next('S')
-	cn.send(b)
+	if err := cn.send(b); err != nil {
+		return nil, cn.errHandleFatalNetErrors(err)
+	}
 
 	cn.readParseResponse()
 	st.paramTyps, st.colNames, st.colTyps = cn.readStatementDescribeResponse()
 	st.colFmts, st.colFmtData = decideColumnFormats(st.colTyps, cn.disablePreparedBinaryResult)
 	cn.readReadyForQuery()
-	return st
+
+	return st, nil
 }
 
 func (cn *conn) Prepare(q string) (_ driver.Stmt, err error) {
@@ -787,7 +795,7 @@ func (cn *conn) Prepare(q string) (_ driver.Stmt, err error) {
 		}
 		return s, err
 	}
-	return cn.prepareTo(q, cn.gname()), nil
+	return cn.prepareTo(q, cn.gname())
 }
 
 func (cn *conn) Close() (err error) {
@@ -829,7 +837,9 @@ func (cn *conn) query(query string, args []driver.Value) (_ *rows, err error) {
 	}
 
 	if cn.binaryParameters {
-		cn.sendBinaryModeQuery(query, args)
+		if err := cn.sendBinaryModeQuery(query, args); err != nil {
+			return nil, cn.errHandleFatalNetErrors(err)
+		}
 
 		cn.readParseResponse()
 		cn.readBindResponse()
@@ -838,7 +848,11 @@ func (cn *conn) query(query string, args []driver.Value) (_ *rows, err error) {
 		cn.postExecuteWorkaround()
 		return rows, nil
 	}
-	st := cn.prepareTo(query, "")
+	st, err := cn.prepareTo(query, "")
+	if err != nil {
+		return nil, err
+	}
+
 	st.exec(args)
 	return &rows{
 		cn:         cn,
@@ -862,7 +876,9 @@ func (cn *conn) Exec(query string, args []driver.Value) (res driver.Result, err
 	}
 
 	if cn.binaryParameters {
-		cn.sendBinaryModeQuery(query, args)
+		if err := cn.sendBinaryModeQuery(query, args); err != nil {
+			return res, cn.errHandleFatalNetErrors(err)
+		}
 
 		cn.readParseResponse()
 		cn.readBindResponse()
@@ -874,16 +890,25 @@ func (cn *conn) Exec(query string, args []driver.Value) (res driver.Result, err
 	// Use the unnamed statement to defer planning until bind
 	// time, or else value-based selectivity estimates cannot be
 	// used.
-	st := cn.prepareTo(query, "")
+	st, err := cn.prepareTo(query, "")
+	if err != nil {
+		return res, nil
+	}
+
 	r, err := st.Exec(args)
 	if err != nil {
 		panic(err)
 	}
 	return r, err
 }
 
-func (cn *conn) send(m *writeBuf) {
+func (cn *conn) send(m *writeBuf) error {
 	_, err := cn.c.Write(m.wrap())
+	return err
+}
+
+func (cn *conn) mustSend(m *writeBuf) {
+	err := cn.send(m)
 	if err != nil {
 		panic(err)
 	}
@@ -1109,7 +1134,7 @@ func (cn *conn) auth(r *readBuf, o values) {
 	case 3:
 		w := cn.writeBuf('p')
 		w.string(o["password"])
-		cn.send(w)
+		cn.mustSend(w)
 
 		t, r := cn.recv()
 		if t != 'R' {
@@ -1123,7 +1148,7 @@ func (cn *conn) auth(r *readBuf, o values) {
 		s := string(r.next(4))
 		w := cn.writeBuf('p')
 		w.string("md5" + md5s(md5s(o["password"]+o["user"])+s))
-		cn.send(w)
+		cn.mustSend(w)
 
 		t, r := cn.recv()
 		if t != 'R' {
@@ -1145,7 +1170,7 @@ func (cn *conn) auth(r *readBuf, o values) {
 		w.string("SCRAM-SHA-256")
 		w.int32(len(scOut))
 		w.bytes(scOut)
-		cn.send(w)
+		cn.mustSend(w)
 
 		t, r := cn.recv()
 		if t != 'R' {
@@ -1165,7 +1190,7 @@ func (cn *conn) auth(r *readBuf, o values) {
 		scOut = sc.Out()
 		w = cn.writeBuf('p')
 		w.bytes(scOut)
-		cn.send(w)
+		cn.mustSend(w)
 
 		t, r = cn.recv()
 		if t != 'R' {
@@ -1219,9 +1244,11 @@ func (st *stmt) Close() (err error) {
 	w := st.cn.writeBuf('C')
 	w.byte('S')
 	w.string(st.name)
-	st.cn.send(w)
+	if err := st.cn.send(w); err != nil {
+		return st.cn.errHandleFatalNetErrors(err)
+	}
 
-	st.cn.send(st.cn.writeBuf('S'))
+	st.cn.mustSend(st.cn.writeBuf('S'))
 
 	t, _ := st.cn.recv1()
 	if t != '3' {
@@ -1299,7 +1326,7 @@ func (st *stmt) exec(v []driver.Value) {
 	w.int32(0)
 
 	w.next('S')
-	cn.send(w)
+	cn.mustSend(w)
 
 	cn.readBindResponse()
 	cn.postExecuteWorkaround()
@@ -1577,7 +1604,7 @@ func (cn *conn) sendBinaryParameters(b *writeBuf, args []driver.Value) {
 	}
 }
 
-func (cn *conn) sendBinaryModeQuery(query string, args []driver.Value) {
+func (cn *conn) sendBinaryModeQuery(query string, args []driver.Value) error {
 	if len(args) >= 65536 {
 		errorf("got %d parameters but PostgreSQL only supports 65535 parameters", len(args))
 	}
@@ -1601,7 +1628,7 @@ func (cn *conn) sendBinaryModeQuery(query string, args []driver.Value) {
 	b.int32(0)
 
 	b.next('S')
-	cn.send(b)
+	return cn.send(b)
 }
 
 func (cn *conn) processParameterStatus(r *readBuf) {

error.go

@@ -480,36 +480,56 @@ func errRecoverNoErrBadConn(err *error) {
 
 func (c *conn) errRecover(err *error) {
 	e := recover()
-	switch v := e.(type) {
-	case nil:
-		// Do nothing
+	if e == nil {
+		return
+	}
+
+	if val, ok := e.(error); ok {
+		*err = c.errHandleFatalNetErrors(val)
+		return
+	}
+
+	c.bad = true
+	panic(fmt.Sprintf("unknown error: %#v", e))
+}
+
+func (c *conn) errHandleFatalNetErrors(e error) error {
+	return c.errHandle(e, false)
+}
+
+func (c *conn) errHandle(e error, netErrorIsRetryable bool) error {
+	var result error
+
+	switch v := (e).(type) {
 	case runtime.Error:
 		c.bad = true
 		panic(v)
 	case *Error:
 		if v.Fatal() {
-			*err = driver.ErrBadConn
+			result = driver.ErrBadConn
 		} else {
-			*err = v
+			result = v
 		}
 	case *net.OpError:
 		c.bad = true
-		*err = v
-	case error:
-		if v == io.EOF || v.(error).Error() == "remote error: handshake failure" {
-			*err = driver.ErrBadConn
+		if netErrorIsRetryable {
+			result = driver.ErrBadConn
 		} else {
-			*err = v
+			result = v
 		}
-
 	default:
-		c.bad = true
-		panic(fmt.Sprintf("unknown error: %#v", e))
+		if v == io.EOF || v.(error).Error() == "remote error: handshake failure" {
+			result = driver.ErrBadConn
+		} else {
+			result = v
+		}
 	}
 
 	// Any time we return ErrBadConn, we need to remember it since *Tx doesn't
 	// mark the connection bad in database/sql.
-	if *err == driver.ErrBadConn {
+	if result == driver.ErrBadConn {
 		c.bad = true
 	}
+
+	return result
 }