Sync flow: use metadata for last offset fetch and cancel groupCtx

[Amogh-Bharadwaj]

Consider the following scenario where you have a mirror to ClickHouse with 2 replicas:

1. Normalize takes a while due to wide tables on ClickHouse with some MVs on them and so let's say normalize is 10 batches behind sync flow.
2. Normalize is currently loading 10 batches of avro files into the raw table, and these queries run on replica A.
3. While this is happening, sync flow tries to open a destination connection to call destinationConnector.GetLastOffset, but we try to open a connection to replica B.
4. Let's say at the time we do the above 3rd step, replica B is down for some reason

In this case, sync flow never retries because normalizeLoop is stuck on normalizing batches and so does not get the chance to see that the syncDone channel is closed (the closing is done when sync flow errors out, in this case due to last offset erroring)
And so we get stuck on the errgroup waiting on normalize to finish -- so we never retry sync flow in this case

This PR:

1. Instead calls metadata.GetLastOffset rather than calling this method through connector construction which opens connections to the data stores -- an unneeded step when we just want to get something from catalog
2. Makes groupCtx cancelled if sync flow errors out. The two goroutines which rely on it are:
   i) maintainReplConn which does not error out if groupCtx is cancelled
   ii) normalizeLoop which anyway will exit because of syncDone

The whole purpose of cancelling groupCtx when syncErr != nil is so that we hit this branch which leads to syncflow activity retrying:

			if groupCtx.Err() != nil {
				// need to return ctx.Err(), avoid returning syncErr that's wrapped context canceled
				break
			}

[Copilot]

Pull Request Overview

This PR refactors how the sync flow retrieves its last offset (using a metadata-only client rather than opening full DB connectors) and ensures the errgroup’s context is cancelled on sync errors so retries can proceed.

• Switch offset fetch to use external_metadata client instead of opening source/destination connectors
• Wrap the errgroup in a cancellable context and call cancelFunc() when sync errors to trigger retries

Reviewed Changes

Copilot reviewed 2 out of 2 changed files in this pull request and generated no comments.

File	Description
flow/activities/flowable_core.go	Replaced connector-based GetLastOffset calls with metadata
flow/activities/flowable.go	Introduced context.WithCancel for the errgroup and invoke cancellation on sync failure

Comments suppressed due to low confidence (2)

flow/activities/flowable.go:387

• The maintainReplConn goroutine may not return on context cancellation, causing errgroup.Wait() to hang and preventing retries. Ensure maintainReplConn watches groupCtx.Done() and returns promptly when the context is cancelled.

			cancelFunc()

flow/activities/flowable_core.go:143

• [nitpick] Confirm whether the metadata client returned by NewPostgresMetadataFromCatalog holds resources that need explicit closing (e.g., a DB connection). If so, add a defer pgMetadata.Close() or equivalent to avoid resource leaks.

		return pgMetadata.GetLastOffset(ctx, flowName)

[serprex]

this doesn't work correctly for pg to pg replication where we store metadata on destination peer instead of catalog

I've brought up wanting to change this, but for now you need to check if source & destination are both pg

In most cases you can return GetLastOffset from srcConn which'll usually do what you're doing here. The problem is that originally metadata was managed by destination connector

[serprex]

seems like we should be running the sync loop as another waitGroup.Go so it can return syncErr to cancel the errgroup