fix: retry on deadlock

db.go

@@ -267,3 +267,12 @@ func IsForeignKeyError(err error) bool {
 
 	return false
 }
+
+func IsDeadlockError(err error) bool {
+	var pgErr *pgconn.PgError
+	if errors.As(err, &pgErr) {
+		return pgErr.Code == pgerrcode.DeadlockDetected
+	}
+
+	return false
+}

go.mod

@@ -37,6 +37,7 @@ require (
 	github.com/robfig/cron/v3 v3.0.1
 	github.com/rodaine/table v1.1.0
 	github.com/samber/lo v1.39.0
+	github.com/sethvargo/go-retry v0.2.4
 	github.com/spf13/pflag v1.0.5
 	github.com/timberio/go-datemath v0.1.0
 	github.com/xeipuuv/gojsonschema v1.2.0

go.sum

@@ -1345,6 +1345,8 @@ github.com/samber/lo v1.39.0/go.mod h1:+m/ZKRl6ClXCE2Lgf3MsQlWfh4bn1bz6CXEOxnEXn
 github.com/sean-/seed v0.0.0-20170313163322-e2103e2c3529/go.mod h1:DxrIzT+xaE7yg65j358z/aeFdxmN0P9QXhEzd20vsDc=
 github.com/sergi/go-diff v1.3.1 h1:xkr+Oxo4BOQKmkn/B9eMK0g5Kg/983T9DqqPHwYqD+8=
 github.com/sergi/go-diff v1.3.1/go.mod h1:aMJSSKb2lpPvRNec0+w3fl7LP9IOFzdc9Pa4NFbPK1I=
+github.com/sethvargo/go-retry v0.2.4 h1:T+jHEQy/zKJf5s95UkguisicE0zuF9y7+/vgz08Ocec=
+github.com/sethvargo/go-retry v0.2.4/go.mod h1:1afjQuvh7s4gflMObvjLPaWgluLLyhA1wmVZ6KLpICw=
 github.com/shurcooL/sanitized_anchor_name v1.0.0/go.mod h1:1NzhyTcUVG4SuEtjjoZeVRXNmyL/1OwPU0+IJeTBvfc=
 github.com/sirupsen/logrus v1.2.0/go.mod h1:LxeOpSwHxABJmUn/MG1IvRgCAasNZTLOkJPxbbu5VWo=
 github.com/sirupsen/logrus v1.4.2/go.mod h1:tLMulIdttU9McNUspp0xgXVQah82FyeX6MwdIuYE2rE=

upstream/jobs.go

@@ -1,13 +1,17 @@
 package upstream
 
 import (
+	gocontext "context"
 	"encoding/json"
 	"fmt"
+	"time"
 
+	"github.com/flanksource/duty"
 	"github.com/flanksource/duty/api"
 	"github.com/flanksource/duty/context"
 	"github.com/flanksource/duty/models"
 	"github.com/samber/lo"
+	"github.com/sethvargo/go-retry"
 	"gorm.io/gorm"
 )
 
@@ -134,14 +138,35 @@ func reconcileTable(ctx context.Context, config UpstreamConfig, table pushableTa
 
 		count += len(items)
 
-		if c, ok := table.(customIsPushedUpdater); ok {
-			if err := c.UpdateIsPushed(ctx.DB(), items); err != nil {
-				return 0, fmt.Errorf("failed to update is_pushed for %s: %w", table.TableName(), err)
-			}
-		} else {
-			ids := lo.Map(items, func(a models.DBTable, _ int) string { return a.PK() })
-			if err := ctx.DB().Model(table).Where("id IN ?", ids).Update("is_pushed", true).Error; err != nil {
-				return 0, fmt.Errorf("failed to update is_pushed on %s: %w", table.TableName(), err)
+		batchSize := ctx.Properties().Int("update_is_pushed.batch.size", 200)
+		for _, batch := range lo.Chunk(items, batchSize) {
+			backoff := retry.WithJitter(time.Second, retry.WithMaxRetries(3, retry.NewExponential(time.Second)))
+			err = retry.Do(ctx, backoff, func(_ctx gocontext.Context) error {
+				ctx = _ctx.(context.Context)
+
+				if c, ok := table.(customIsPushedUpdater); ok {
+					if err := c.UpdateIsPushed(ctx.DB(), batch); err != nil {
+						if duty.IsDeadlockError(err) {
+							return retry.RetryableError(err)
+						}
+
+						return fmt.Errorf("failed to update is_pushed on %s: %w", table.TableName(), err)
+					}
+				} else {
+					ids := lo.Map(batch, func(a models.DBTable, _ int) string { return a.PK() })
+					if err := ctx.DB().Model(table).Where("id IN ?", ids).Update("is_pushed", true).Error; err != nil {
+						if duty.IsDeadlockError(err) {
+							return retry.RetryableError(err)
+						}
+
+						return fmt.Errorf("failed to update is_pushed on %s: %w", table.TableName(), err)
+					}
+				}
+
+				return nil
+			})
+			if err != nil {
+				return count, err
 			}
 		}