From c77f2fd6f47abe61bf25349c5338a9caa884ab5c Mon Sep 17 00:00:00 2001 From: Ilia Demianenko Date: Fri, 4 Sep 2026 18:42:05 -0600 Subject: [PATCH 1/5] Remove extra lock and ping from catalog pool after init --- flow/internal/catalog.go | 51 ++++++++++++++++++++++------------------ 1 file changed, 28 insertions(+), 23 deletions(-) diff --git a/flow/internal/catalog.go b/flow/internal/catalog.go index 8c3aa45444..d59a6f2799 100644 --- a/flow/internal/catalog.go +++ b/flow/internal/catalog.go @@ -4,6 +4,7 @@ import ( "context" "fmt" "sync" + "sync/atomic" "time" "github.com/jackc/pgx/v5/pgxpool" @@ -15,35 +16,39 @@ import ( var ( poolMutex = &sync.Mutex{} - pool *pgxpool.Pool + pool atomic.Pointer[pgxpool.Pool] ) func GetCatalogConnectionPoolFromEnv(ctx context.Context) (shared.CatalogPool, error) { - poolMutex.Lock() - defer poolMutex.Unlock() - if pool == nil { - var err error - catalogConnectionString := GetCatalogConnectionStringFromEnv(ctx) - config, err := pgxpool.ParseConfig(catalogConnectionString) - if err != nil { - return shared.CatalogPool{}, - exceptions.NewCatalogError(fmt.Errorf("unable to parse catalog connection string: %w", err)) - } - config.MaxConns = 3 - config.MaxConnIdleTime = 90 * time.Second - pool, err = pgxpool.NewWithConfig(ctx, config) - if err != nil { - return shared.CatalogPool{Pool: pool}, - exceptions.NewCatalogError(fmt.Errorf("unable to establish connection with catalog: %w", err)) - } - } + if pool.Load() == nil { + poolMutex.Lock() + defer poolMutex.Unlock() + if pool.Load() == nil { + var err error + catalogConnectionString := GetCatalogConnectionStringFromEnv(ctx) + config, err := pgxpool.ParseConfig(catalogConnectionString) + if err != nil { + return shared.CatalogPool{}, + exceptions.NewCatalogError(fmt.Errorf("unable to parse catalog connection string: %w", err)) + } + config.MaxConns = 3 + config.MaxConnIdleTime = 90 * time.Second + localPool, err := pgxpool.NewWithConfig(ctx, config) + if err != nil { + return shared.CatalogPool{}, + exceptions.NewCatalogError(fmt.Errorf("unable to initialize catalog connection pool: %w", err)) + } - if err := pool.Ping(ctx); err != nil { - return shared.CatalogPool{Pool: pool}, - exceptions.NewCatalogError(fmt.Errorf("unable to establish connection with catalog: %w", err)) + if err := localPool.Ping(ctx); err != nil { + localPool.Close() + return shared.CatalogPool{}, + exceptions.NewCatalogError(fmt.Errorf("unable to establish connection with catalog: %w", err)) + } + pool.Store(localPool) + } } - return shared.CatalogPool{Pool: pool}, nil + return shared.CatalogPool{Pool: pool.Load()}, nil } func GetCatalogConnectionStringFromEnv(ctx context.Context) string { From 1fbdd368b5c430e6d29eea25e02145239d903fdc Mon Sep 17 00:00:00 2001 From: Ilia Demianenko Date: Fri, 4 Sep 2026 18:44:12 -0600 Subject: [PATCH 2/5] Cache dynconf settings polled per snapshot partition --- flow/connectors/clickhouse/avro_sync.go | 2 +- flow/connectors/clickhouse/clickhouse.go | 15 ++- flow/connectors/clickhouse/staging_s3.go | 7 +- flow/internal/dynamicconf_cache.go | 46 +++++++++ flow/internal/dynamicconf_cache_test.go | 121 +++++++++++++++++++++++ 5 files changed, 184 insertions(+), 7 deletions(-) create mode 100644 flow/internal/dynamicconf_cache.go create mode 100644 flow/internal/dynamicconf_cache_test.go diff --git a/flow/connectors/clickhouse/avro_sync.go b/flow/connectors/clickhouse/avro_sync.go index 47c37304bf..fe66e13267 100644 --- a/flow/connectors/clickhouse/avro_sync.go +++ b/flow/connectors/clickhouse/avro_sync.go @@ -156,7 +156,7 @@ func (s *ClickHouseAvroSyncMethod) pushDataToStagingForSnapshot( return nil, 0, err } - bytesPerAvroFile, err := internal.PeerDBS3BytesPerAvroFile(ctx, config.Env) + bytesPerAvroFile, err := s.bytesPerAvroFile.Get(ctx, config.Env) if err != nil { return nil, 0, err } diff --git a/flow/connectors/clickhouse/clickhouse.go b/flow/connectors/clickhouse/clickhouse.go index a1859baf4f..584aacf5c0 100644 --- a/flow/connectors/clickhouse/clickhouse.go +++ b/flow/connectors/clickhouse/clickhouse.go @@ -29,11 +29,12 @@ import ( type ClickHouseConnector struct { *metadataStore.PostgresMetadata - database clickhouse.Conn - logger log.Logger - Config *protos.ClickhouseConfig - staging StagingStore - chVersion *clickhouseproto.Version + database clickhouse.Conn + logger log.Logger + Config *protos.ClickhouseConfig + staging StagingStore + chVersion *clickhouseproto.Version + bytesPerAvroFile *internal.CachedDynconfSetting[int64] } func NewClickHouseConnector( @@ -70,6 +71,10 @@ func NewClickHouseConnector( logger: logger, staging: staging, chVersion: &clickHouseVersion.Version, + bytesPerAvroFile: internal.NewCachedDynconfSetting( + internal.PeerDBS3BytesPerAvroFile, + 30*time.Second, + ), }, nil } diff --git a/flow/connectors/clickhouse/staging_s3.go b/flow/connectors/clickhouse/staging_s3.go index 970ed9cbc1..ee467d739e 100644 --- a/flow/connectors/clickhouse/staging_s3.go +++ b/flow/connectors/clickhouse/staging_s3.go @@ -29,6 +29,7 @@ type s3StagingStore struct { bucket string prefix string fullPath string // original "s3://bucket/prefix" for logging + partSize *internal.CachedDynconfSetting[int64] } //nolint:iface // factory function intentionally returns interface @@ -98,6 +99,10 @@ func newS3StagingStore( prefix: s3o.Prefix, fullPath: awsBucketPath, creds: credentialsProvider, + partSize: internal.NewCachedDynconfSetting( + internal.PeerDBS3PartSize, + 30*time.Second, + ), }, nil } @@ -109,7 +114,7 @@ func (s *s3StagingStore) Upload(ctx context.Context, env map[string]string, key return fmt.Errorf("failed to create S3 client: %w", err) } - partSize, err := internal.PeerDBS3PartSize(ctx, env) + partSize, err := s.partSize.Get(ctx, env) if err != nil { return fmt.Errorf("could not get s3 part size config: %w", err) } diff --git a/flow/internal/dynamicconf_cache.go b/flow/internal/dynamicconf_cache.go new file mode 100644 index 0000000000..d7b909a96d --- /dev/null +++ b/flow/internal/dynamicconf_cache.go @@ -0,0 +1,46 @@ +package internal + +import ( + "context" + "sync" + "time" +) + +// CachedDynconfSetting caches the value returned by a typed dynamic setting getter. +type CachedDynconfSetting[T any] struct { + getter func(context.Context, map[string]string) (T, error) + ttl time.Duration + mu sync.Mutex + value T + loadedAt time.Time +} + +// NewCachedDynconfSetting creates a typed dynamic setting cache with the given lifetime. +func NewCachedDynconfSetting[T any]( + getter func(context.Context, map[string]string) (T, error), + ttl time.Duration, +) *CachedDynconfSetting[T] { + return &CachedDynconfSetting[T]{ + getter: getter, + ttl: ttl, + } +} + +// Get returns the cached value or refreshes it using the configured getter. +func (s *CachedDynconfSetting[T]) Get(ctx context.Context, env map[string]string) (T, error) { + s.mu.Lock() + defer s.mu.Unlock() + + if !s.loadedAt.IsZero() && time.Since(s.loadedAt) < s.ttl { + return s.value, nil + } + + value, err := s.getter(ctx, env) + if err != nil { + var zero T + return zero, err + } + s.value = value + s.loadedAt = time.Now() + return value, nil +} diff --git a/flow/internal/dynamicconf_cache_test.go b/flow/internal/dynamicconf_cache_test.go new file mode 100644 index 0000000000..1ad1b5c855 --- /dev/null +++ b/flow/internal/dynamicconf_cache_test.go @@ -0,0 +1,121 @@ +package internal + +import ( + "context" + "errors" + "math/rand/v2" + "strconv" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/stretchr/testify/require" +) + +func TestCachedDynconfSettingCachesWithinTTL(t *testing.T) { + var calls int + setting := NewCachedDynconfSetting(func(context.Context, map[string]string) (int64, error) { + calls++ + return 10, nil + }, time.Minute) + + for range 2 { + value, err := setting.Get(t.Context(), nil) + require.NoError(t, err) + require.Equal(t, int64(10), value) + } + require.Equal(t, 1, calls) +} + +func TestCachedDynconfSettingRefreshesAfterTTL(t *testing.T) { + var calls int64 + setting := NewCachedDynconfSetting(func(context.Context, map[string]string) (int64, error) { + calls++ + return calls * 10, nil + }, time.Millisecond) + + value, err := setting.Get(t.Context(), nil) + require.NoError(t, err) + require.Equal(t, int64(10), value) + + time.Sleep(10 * time.Millisecond) + + value, err = setting.Get(t.Context(), nil) + require.NoError(t, err) + require.Equal(t, int64(20), value) + require.Equal(t, int64(2), calls) +} + +func TestCachedDynconfSettingDoesNotCacheErrors(t *testing.T) { + wantErr := errors.New("lookup failed") + var calls int + setting := NewCachedDynconfSetting(func(context.Context, map[string]string) (int64, error) { + calls++ + if calls == 1 { + return 0, wantErr + } + return 10, nil + }, time.Minute) + + _, err := setting.Get(t.Context(), nil) + require.ErrorIs(t, err, wantErr) + value, err := setting.Get(t.Context(), nil) + require.NoError(t, err) + require.Equal(t, int64(10), value) + require.Equal(t, 2, calls) +} + +func TestCachedDynconfSettingCoalescesConcurrentRefreshes(t *testing.T) { + const goroutines = 8 + + var calls atomic.Int32 + started := make(chan struct{}) + release := make(chan struct{}) + setting := NewCachedDynconfSetting(func(context.Context, map[string]string) (int64, error) { + if calls.Add(1) == 1 { + close(started) + } + <-release + return 10, nil + }, time.Minute) + + results := make(chan int64, goroutines) + errs := make(chan error, goroutines) + var wg sync.WaitGroup + wg.Add(goroutines) + for range goroutines { + go func() { + defer wg.Done() + value, err := setting.Get(t.Context(), nil) + results <- value + errs <- err + }() + } + + <-started + close(release) + wg.Wait() + close(results) + close(errs) + + for err := range errs { + require.NoError(t, err) + } + for value := range results { + require.Equal(t, int64(10), value) + } + require.Equal(t, int32(1), calls.Load()) +} + +func TestCachedDynconfSettingWithTypedDynconfGetter(t *testing.T) { + //nolint:gosec // Test data does not need cryptographically secure randomness. + expectedPartSize := rand.Int64() + setting := NewCachedDynconfSetting(PeerDBS3PartSize, time.Minute) + + value, err := setting.Get(t.Context(), map[string]string{ + "PEERDB_S3_PART_SIZE": strconv.FormatInt(expectedPartSize, 10), + }) + require.NoError(t, err) + require.Equal(t, expectedPartSize, value) +} From 35427fad0a39aec2be6968251ffd135b660fde49 Mon Sep 17 00:00:00 2001 From: Ilia Demianenko Date: Fri, 4 Sep 2026 20:29:56 -0600 Subject: [PATCH 3/5] Global cache per setting --- flow/connectors/clickhouse/avro_sync.go | 2 +- flow/connectors/clickhouse/clickhouse.go | 15 ++---- flow/connectors/clickhouse/staging_s3.go | 7 +-- flow/internal/dynamicconf.go | 10 +++- flow/internal/dynamicconf_cache.go | 38 ++++++++++----- flow/internal/dynamicconf_cache_test.go | 61 ++++++++++++++++++++---- 6 files changed, 93 insertions(+), 40 deletions(-) diff --git a/flow/connectors/clickhouse/avro_sync.go b/flow/connectors/clickhouse/avro_sync.go index fe66e13267..47c37304bf 100644 --- a/flow/connectors/clickhouse/avro_sync.go +++ b/flow/connectors/clickhouse/avro_sync.go @@ -156,7 +156,7 @@ func (s *ClickHouseAvroSyncMethod) pushDataToStagingForSnapshot( return nil, 0, err } - bytesPerAvroFile, err := s.bytesPerAvroFile.Get(ctx, config.Env) + bytesPerAvroFile, err := internal.PeerDBS3BytesPerAvroFile(ctx, config.Env) if err != nil { return nil, 0, err } diff --git a/flow/connectors/clickhouse/clickhouse.go b/flow/connectors/clickhouse/clickhouse.go index 584aacf5c0..a1859baf4f 100644 --- a/flow/connectors/clickhouse/clickhouse.go +++ b/flow/connectors/clickhouse/clickhouse.go @@ -29,12 +29,11 @@ import ( type ClickHouseConnector struct { *metadataStore.PostgresMetadata - database clickhouse.Conn - logger log.Logger - Config *protos.ClickhouseConfig - staging StagingStore - chVersion *clickhouseproto.Version - bytesPerAvroFile *internal.CachedDynconfSetting[int64] + database clickhouse.Conn + logger log.Logger + Config *protos.ClickhouseConfig + staging StagingStore + chVersion *clickhouseproto.Version } func NewClickHouseConnector( @@ -71,10 +70,6 @@ func NewClickHouseConnector( logger: logger, staging: staging, chVersion: &clickHouseVersion.Version, - bytesPerAvroFile: internal.NewCachedDynconfSetting( - internal.PeerDBS3BytesPerAvroFile, - 30*time.Second, - ), }, nil } diff --git a/flow/connectors/clickhouse/staging_s3.go b/flow/connectors/clickhouse/staging_s3.go index ee467d739e..970ed9cbc1 100644 --- a/flow/connectors/clickhouse/staging_s3.go +++ b/flow/connectors/clickhouse/staging_s3.go @@ -29,7 +29,6 @@ type s3StagingStore struct { bucket string prefix string fullPath string // original "s3://bucket/prefix" for logging - partSize *internal.CachedDynconfSetting[int64] } //nolint:iface // factory function intentionally returns interface @@ -99,10 +98,6 @@ func newS3StagingStore( prefix: s3o.Prefix, fullPath: awsBucketPath, creds: credentialsProvider, - partSize: internal.NewCachedDynconfSetting( - internal.PeerDBS3PartSize, - 30*time.Second, - ), }, nil } @@ -114,7 +109,7 @@ func (s *s3StagingStore) Upload(ctx context.Context, env map[string]string, key return fmt.Errorf("failed to create S3 client: %w", err) } - partSize, err := s.partSize.Get(ctx, env) + partSize, err := internal.PeerDBS3PartSize(ctx, env) if err != nil { return fmt.Errorf("could not get s3 part size config: %w", err) } diff --git a/flow/internal/dynamicconf.go b/flow/internal/dynamicconf.go index 57d0d7e260..82e855fd59 100644 --- a/flow/internal/dynamicconf.go +++ b/flow/internal/dynamicconf.go @@ -862,12 +862,18 @@ func PeerDBS3UuidPrefix(ctx context.Context, env map[string]string) (bool, error return dynamicConfBool(ctx, env, "PEERDB_S3_UUID_PREFIX") } +var peerDBS3PartSizeCache CachedDynconfSetting[int64] + func PeerDBS3PartSize(ctx context.Context, env map[string]string) (int64, error) { - return dynamicConfSigned[int64](ctx, env, "PEERDB_S3_PART_SIZE") + peerDBS3PartSizeCache.InitOnce("PEERDB_S3_PART_SIZE", 30*time.Second, dynamicConfSigned[int64]) + return peerDBS3PartSizeCache.Get(ctx, env) } +var peerDBS3BytesPerAvroFileCache CachedDynconfSetting[int64] + func PeerDBS3BytesPerAvroFile(ctx context.Context, env map[string]string) (int64, error) { - return dynamicConfSigned[int64](ctx, env, "PEERDB_S3_BYTES_PER_AVRO_FILE") + peerDBS3BytesPerAvroFileCache.InitOnce("PEERDB_S3_BYTES_PER_AVRO_FILE", 30*time.Second, dynamicConfSigned[int64]) + return peerDBS3BytesPerAvroFileCache.Get(ctx, env) } // Kafka has topic auto create as an option, auto.create.topics.enable diff --git a/flow/internal/dynamicconf_cache.go b/flow/internal/dynamicconf_cache.go index d7b909a96d..5c3b079687 100644 --- a/flow/internal/dynamicconf_cache.go +++ b/flow/internal/dynamicconf_cache.go @@ -2,32 +2,46 @@ package internal import ( "context" + "errors" "sync" "time" ) // CachedDynconfSetting caches the value returned by a typed dynamic setting getter. type CachedDynconfSetting[T any] struct { - getter func(context.Context, map[string]string) (T, error) + loadedAt time.Time + value T + getter func(context.Context, map[string]string, string) (T, error) + name string ttl time.Duration + once sync.Once mu sync.Mutex - value T - loadedAt time.Time } -// NewCachedDynconfSetting creates a typed dynamic setting cache with the given lifetime. -func NewCachedDynconfSetting[T any]( - getter func(context.Context, map[string]string) (T, error), +// InitOnce configures the setting cache on its first call. +func (s *CachedDynconfSetting[T]) InitOnce( + name string, ttl time.Duration, -) *CachedDynconfSetting[T] { - return &CachedDynconfSetting[T]{ - getter: getter, - ttl: ttl, - } + getter func(context.Context, map[string]string, string) (T, error), +) { + s.once.Do(func() { + s.name = name + s.ttl = ttl + s.getter = getter + }) } // Get returns the cached value or refreshes it using the configured getter. func (s *CachedDynconfSetting[T]) Get(ctx context.Context, env map[string]string) (T, error) { + if s.getter == nil { + var zero T + return zero, errors.New("cached dynamic setting is not initialized") + } + + if _, overridden := env[s.name]; overridden { + return s.getter(ctx, env, s.name) + } + s.mu.Lock() defer s.mu.Unlock() @@ -35,7 +49,7 @@ func (s *CachedDynconfSetting[T]) Get(ctx context.Context, env map[string]string return s.value, nil } - value, err := s.getter(ctx, env) + value, err := s.getter(ctx, nil, s.name) if err != nil { var zero T return zero, err diff --git a/flow/internal/dynamicconf_cache_test.go b/flow/internal/dynamicconf_cache_test.go index 1ad1b5c855..6eb1401926 100644 --- a/flow/internal/dynamicconf_cache_test.go +++ b/flow/internal/dynamicconf_cache_test.go @@ -15,10 +15,12 @@ import ( func TestCachedDynconfSettingCachesWithinTTL(t *testing.T) { var calls int - setting := NewCachedDynconfSetting(func(context.Context, map[string]string) (int64, error) { + getter := func(context.Context, map[string]string, string) (int64, error) { calls++ return 10, nil - }, time.Minute) + } + var setting CachedDynconfSetting[int64] + setting.InitOnce("TEST_SETTING", time.Minute, getter) for range 2 { value, err := setting.Get(t.Context(), nil) @@ -28,12 +30,21 @@ func TestCachedDynconfSettingCachesWithinTTL(t *testing.T) { require.Equal(t, 1, calls) } +func TestCachedDynconfSettingRequiresInitialization(t *testing.T) { + var setting CachedDynconfSetting[int64] + + _, err := setting.Get(t.Context(), nil) + require.EqualError(t, err, "cached dynamic setting is not initialized") +} + func TestCachedDynconfSettingRefreshesAfterTTL(t *testing.T) { var calls int64 - setting := NewCachedDynconfSetting(func(context.Context, map[string]string) (int64, error) { + getter := func(context.Context, map[string]string, string) (int64, error) { calls++ return calls * 10, nil - }, time.Millisecond) + } + var setting CachedDynconfSetting[int64] + setting.InitOnce("TEST_SETTING", time.Millisecond, getter) value, err := setting.Get(t.Context(), nil) require.NoError(t, err) @@ -50,13 +61,15 @@ func TestCachedDynconfSettingRefreshesAfterTTL(t *testing.T) { func TestCachedDynconfSettingDoesNotCacheErrors(t *testing.T) { wantErr := errors.New("lookup failed") var calls int - setting := NewCachedDynconfSetting(func(context.Context, map[string]string) (int64, error) { + getter := func(context.Context, map[string]string, string) (int64, error) { calls++ if calls == 1 { return 0, wantErr } return 10, nil - }, time.Minute) + } + var setting CachedDynconfSetting[int64] + setting.InitOnce("TEST_SETTING", time.Minute, getter) _, err := setting.Get(t.Context(), nil) require.ErrorIs(t, err, wantErr) @@ -72,13 +85,14 @@ func TestCachedDynconfSettingCoalescesConcurrentRefreshes(t *testing.T) { var calls atomic.Int32 started := make(chan struct{}) release := make(chan struct{}) - setting := NewCachedDynconfSetting(func(context.Context, map[string]string) (int64, error) { + getter := func(context.Context, map[string]string, string) (int64, error) { if calls.Add(1) == 1 { close(started) } <-release return 10, nil - }, time.Minute) + } + var setting CachedDynconfSetting[int64] results := make(chan int64, goroutines) errs := make(chan error, goroutines) @@ -87,6 +101,7 @@ func TestCachedDynconfSettingCoalescesConcurrentRefreshes(t *testing.T) { for range goroutines { go func() { defer wg.Done() + setting.InitOnce("TEST_SETTING", time.Minute, getter) value, err := setting.Get(t.Context(), nil) results <- value errs <- err @@ -111,7 +126,8 @@ func TestCachedDynconfSettingCoalescesConcurrentRefreshes(t *testing.T) { func TestCachedDynconfSettingWithTypedDynconfGetter(t *testing.T) { //nolint:gosec // Test data does not need cryptographically secure randomness. expectedPartSize := rand.Int64() - setting := NewCachedDynconfSetting(PeerDBS3PartSize, time.Minute) + var setting CachedDynconfSetting[int64] + setting.InitOnce("PEERDB_S3_PART_SIZE", time.Minute, dynamicConfSigned[int64]) value, err := setting.Get(t.Context(), map[string]string{ "PEERDB_S3_PART_SIZE": strconv.FormatInt(expectedPartSize, 10), @@ -119,3 +135,30 @@ func TestCachedDynconfSettingWithTypedDynconfGetter(t *testing.T) { require.NoError(t, err) require.Equal(t, expectedPartSize, value) } + +func TestCachedDynconfSettingDoesNotCacheEnvOverrides(t *testing.T) { + const name = "TEST_SETTING" + var calls int + getter := func(_ context.Context, env map[string]string, name string) (int64, error) { + calls++ + if value, overridden := env[name]; overridden { + return strconv.ParseInt(value, 10, 64) + } + return 10, nil + } + var setting CachedDynconfSetting[int64] + setting.InitOnce(name, time.Minute, getter) + + value, err := setting.Get(t.Context(), nil) + require.NoError(t, err) + require.Equal(t, int64(10), value) + + value, err = setting.Get(t.Context(), map[string]string{name: "20"}) + require.NoError(t, err) + require.Equal(t, int64(20), value) + + value, err = setting.Get(t.Context(), nil) + require.NoError(t, err) + require.Equal(t, int64(10), value) + require.Equal(t, 2, calls) +} From 1d5d3917d5e00d1449196889f0c7f487997fac63 Mon Sep 17 00:00:00 2001 From: Ilia Demianenko Date: Fri, 4 Sep 2026 22:11:26 -0600 Subject: [PATCH 4/5] Merge two bookkeeping calls --- flow/activities/flowable_core.go | 12 ++-------- .../connectors/utils/monitoring/monitoring.go | 22 +++++-------------- 2 files changed, 8 insertions(+), 26 deletions(-) diff --git a/flow/activities/flowable_core.go b/flow/activities/flowable_core.go index 8cda09d876..e00b4a62c9 100644 --- a/flow/activities/flowable_core.go +++ b/flow/activities/flowable_core.go @@ -559,12 +559,9 @@ func replicateQRepPartition[TRead any, TWrite QRepStreamCloser, TSync connectors if rowsSynced > 0 { logger.Info(fmt.Sprintf("pushed %d records", rowsSynced)) - if err := monitoring.UpdateRowsSyncedForPartition(ctx, a.CatalogPool, rowsSynced, runUUID, partition); err != nil { - return err - } } - return monitoring.UpdateEndTimeForPartition(ctx, a.CatalogPool, runUUID, partition) + return monitoring.UpdateEndTimeAndRowsSyncedForPartition(ctx, a.CatalogPool, rowsSynced, runUUID, partition) } // replicateXminPartition replicates a XminPartition from the source to the destination. @@ -675,15 +672,10 @@ func replicateXminPartition[TRead any, TWrite QRepStreamCloser, TSync connectors } if rowsSynced > 0 { - err := monitoring.UpdateRowsSyncedForPartition(ctx, a.CatalogPool, rowsSynced, runUUID, partition) - if err != nil { - return 0, err - } - logger.Info(fmt.Sprintf("pushed %d records", rowsSynced)) } - if err := monitoring.UpdateEndTimeForPartition(ctx, a.CatalogPool, runUUID, partition); err != nil { + if err := monitoring.UpdateEndTimeAndRowsSyncedForPartition(ctx, a.CatalogPool, rowsSynced, runUUID, partition); err != nil { return 0, err } diff --git a/flow/connectors/utils/monitoring/monitoring.go b/flow/connectors/utils/monitoring/monitoring.go index ddffb4c5cd..69baa36ba0 100644 --- a/flow/connectors/utils/monitoring/monitoring.go +++ b/flow/connectors/utils/monitoring/monitoring.go @@ -458,26 +458,16 @@ func UpdatePullEndTimeAndRowsForPartition(ctx context.Context, pool shared.Catal return nil } -func UpdateEndTimeForPartition(ctx context.Context, pool shared.CatalogPool, runUUID string, +func UpdateEndTimeAndRowsSyncedForPartition(ctx context.Context, pool shared.CatalogPool, rowsSynced int64, runUUID string, partition *protos.QRepPartition, ) error { if _, err := pool.Exec(ctx, - `UPDATE peerdb_stats.qrep_partitions SET end_time=$1 WHERE run_uuid=$2 AND partition_uuid=$3`, - time.Now(), runUUID, partition.PartitionId, + `UPDATE peerdb_stats.qrep_partitions + SET end_time=$1, rows_synced=CASE WHEN $2::bigint > 0 THEN $2 ELSE rows_synced END + WHERE run_uuid=$3 AND partition_uuid=$4`, + time.Now(), rowsSynced, runUUID, partition.PartitionId, ); err != nil { - return fmt.Errorf("error while updating qrep partition in qrep_partitions: %w", err) - } - return nil -} - -func UpdateRowsSyncedForPartition(ctx context.Context, pool shared.CatalogPool, rowsSynced int64, runUUID string, - partition *protos.QRepPartition, -) error { - if _, err := pool.Exec(ctx, - `UPDATE peerdb_stats.qrep_partitions SET rows_synced=$1 WHERE run_uuid=$2 AND partition_uuid=$3`, - rowsSynced, runUUID, partition.PartitionId, - ); err != nil { - return fmt.Errorf("error while updating rows_synced in qrep_partitions: %w", err) + return fmt.Errorf("error while completing qrep partition in qrep_partitions: %w", err) } return nil } From 60e681dab00ad961249ac88db3bc066759979727 Mon Sep 17 00:00:00 2001 From: Ilia Demianenko Date: Sat, 5 Sep 2026 00:50:14 -0600 Subject: [PATCH 5/5] lint --- flow/internal/dynamicconf_cache_test.go | 1 + 1 file changed, 1 insertion(+) diff --git a/flow/internal/dynamicconf_cache_test.go b/flow/internal/dynamicconf_cache_test.go index 6eb1401926..e83d86c2b7 100644 --- a/flow/internal/dynamicconf_cache_test.go +++ b/flow/internal/dynamicconf_cache_test.go @@ -85,6 +85,7 @@ func TestCachedDynconfSettingCoalescesConcurrentRefreshes(t *testing.T) { var calls atomic.Int32 started := make(chan struct{}) release := make(chan struct{}) + //nolint:unparam // Signature must match the getter type getter := func(context.Context, map[string]string, string) (int64, error) { if calls.Add(1) == 1 { close(started)