Continues & ContinuesTry

misc.go

 package zdb
 
-import "github.com/jackc/pgx/v5"
+import (
+	"github.com/jackc/pgx/v5"
+	"strings"
+)
 
 var ErrNoRows = pgx.ErrNoRows
 
@@ -12,6 +15,16 @@ func remove[T comparable](slice []T, s int) []T {
 	return append(slice[:s], slice[s+1:]...)
 }
 
+func contains(s string, sl []string) bool {
+	for _, s2 := range sl {
+		if strings.Contains(s, s2) {
+			return true
+		}
+	}
+
+	return false
+}
+
 type Logger interface {
 	Printf(format string, v ...any)
 }

pool.go

@@ -31,6 +31,10 @@ type Pool struct {
 	stop bool
 
 	PgTsFormat string
+
+	Continues    []string
+	ContinuesTry []string
+	TryOnError   int
 }
 
 func New() *Pool {
@@ -39,6 +43,9 @@ func New() *Pool {
 		slavesIter:      &atomic.Int64{},
 		slavesAsyncIter: &atomic.Int64{},
 		PgTsFormat:      "2006-01-02 15:04:05",
+		Continues:       []string{"connect", "EOF", "conflict with recovery"},
+		ContinuesTry:    []string{"conflict with recovery"},
+		TryOnError:      1,
 	}
 }
 
@@ -129,6 +136,7 @@ func (d *Pool) async() *conn {
 func (d *Pool) execWrapper(pool connMode, dst any, f func(conn *conn, dst1 any) error) error {
 	for {
 		var q *conn
+		try := 0
 
 		if pool == ConnModeSync {
 			q = d.sync()
@@ -136,11 +144,18 @@ func (d *Pool) execWrapper(pool connMode, dst any, f func(conn *conn, dst1 any)
 			q = d.async()
 		}
 
+	repit:
 		if err := f(q, dst); err != nil {
 			if q.Mode == ConnModeMaster {
 				return err
 			} else {
-				if strings.Contains(err.Error(), "connect") || strings.Contains(err.Error(), "EOF") {
+
+				if try < d.TryOnError && contains(err.Error(), d.ContinuesTry) {
+					try++
+					goto repit
+				}
+
+				if contains(err.Error(), d.Continues) {
 					d.logger.Printf("DB_EXEC_WRAPPER_ERR: %s", err.Error())
 					d.setNotAliveConn(q)
 					continue