feat: disable reconciliation precheck via properties.

context/properties.go

@@ -34,10 +34,9 @@ func (p Properties) Duration(key string, def time.Duration) time.Duration {
 	} else if dur, err := time.ParseDuration(d); err != nil {
 		logger.Warnf("property[%s] invalid duration %s", key, d)
 		return def
-	} else if err == nil {
+	} else {
 		return dur
 	}
-	return def
 }
 
 func (p Properties) Int(key string, def int) int {
@@ -46,10 +45,9 @@ func (p Properties) Int(key string, def int) int {
 	} else if i, err := strconv.Atoi(d); err != nil {
 		logger.Warnf("property[%s] invalid int %s", key, d)
 		return def
-	} else if err == nil {
+	} else {
 		return i
 	}
-	return def
 }
 
 func (p Properties) Off(key string) bool {
@@ -84,7 +82,11 @@ func (k Context) Properties() Properties {
 	return props
 }
 
-func SetLocalProperty(ctx Context, property, value string) {
+func SetLocalProperty(property, value string) {
+	if Local == nil {
+		Local = make(map[string]string)
+	}
+
 	Local[property] = value
 }
 

tests/upstream_test.go

@@ -22,6 +22,8 @@ var _ = ginkgo.Describe("Reconcile Test", ginkgo.Ordered, func() {
 	const agentName = "my-agent"
 
 	ginkgo.BeforeAll(func() {
+		context.SetLocalProperty("upstream.reconcile.pre-check", "false")
+
 		var err error
 		upstreamCtx, drop, err = setup.NewDB(DefaultContext, "upstream")
 		Expect(err).ToNot(HaveOccurred())

upstream/jobs.go

@@ -10,9 +10,6 @@ import (
 	"gorm.io/gorm"
 )
 
-// ReconcilePrecheck, when set, will do an index scan on is_pushed before reconciling
-var ReconcilePrecheck = true
-
 func ReconcileAll(ctx context.Context, config UpstreamConfig, batchSize int) (int, error) {
 	var count int
 
@@ -192,7 +189,7 @@ func reconcileTable[T dbTable](ctx context.Context, config UpstreamConfig, fetch
 	var anon T
 	table := anon.TableName()
 
-	if ReconcilePrecheck {
+	if ctx.Properties()["upstream.reconcile.pre-check"] != "false" {
 		var unpushed float64
 		precheck := fmt.Sprintf(`SELECT reltuples FROM pg_class WHERE relname = '%s_is_pushed_idx'`, table)
 		if err := ctx.DB().Raw(precheck).Scan(&unpushed).Error; err != nil {