Fetch configuration from DB

No longer passing the entire configuration object to the Temporal job as
we can hit the 2MB hard limit imposed by Temporal. Instead, pass
flowJobName and fetch the config from the DB in all the places required.

There are some additional changes that were required to be made, as some
of the jobs are used from multiple contexts - and I had to adapt some of
the information passed to ensure that they continue to work as expected.

Author: alexstoick

docker-compose.yml

@@ -116,7 +116,12 @@ services:
 
   flow-api:
     container_name: flow_api
-    image: ghcr.io/peerdb-io/flow-api:stable-v0.32.1
+    build:
+      context: .
+      dockerfile: stacks/flow.Dockerfile
+      target: flow-api
+      args:
+        PEERDB_VERSION_SHA_SHORT: ${PEERDB_VERSION_SHA_SHORT:-}
     restart: unless-stopped
     ports:
       - 8112:8112
@@ -132,7 +137,10 @@ services:
 
   flow-snapshot-worker:
     container_name: flow-snapshot-worker
-    image: ghcr.io/peerdb-io/flow-snapshot-worker:stable-v0.32.1
+    build:
+      context: .
+      dockerfile: stacks/flow.Dockerfile
+      target: flow-snapshot-worker
     restart: unless-stopped
     environment:
       <<: [*catalog-config, *flow-worker-env, *minio-config]
@@ -142,7 +150,10 @@ services:
 
   flow-worker:
     container_name: flow-worker
-    image: ghcr.io/peerdb-io/flow-worker:stable-v0.32.1
+    build:
+      context: .
+      dockerfile: stacks/flow.Dockerfile
+      target: flow-worker
     restart: unless-stopped
     environment:
       <<: [*catalog-config, *flow-worker-env, *minio-config]

flow/activities/flowable.go

@@ -5,7 +5,9 @@ import (
 	"errors"
 	"fmt"
 	"log/slog"
+	"maps"
 	"os"
+	"slices"
 	"strconv"
 	"sync/atomic"
 	"time"
@@ -124,6 +126,12 @@ func (a *FlowableActivity) EnsurePullability(
 	}
 	defer connectors.CloseConnector(ctx, srcConn)
 
+	cfg, err := internal.FetchConfigFromDB(config.FlowJobName)
+	if err != nil {
+		return nil, a.Alerter.LogFlowError(ctx, config.FlowJobName, fmt.Errorf("failed to fetch config from DB: %w", err))
+	}
+
+	config.SourceTableIdentifiers = slices.Sorted(maps.Keys(internal.TableNameMapping(cfg.TableMappings)))
 	output, err := srcConn.EnsurePullability(ctx, config)
 	if err != nil {
 		return nil, a.Alerter.LogFlowError(ctx, config.FlowJobName, fmt.Errorf("failed to ensure pullability: %w", err))
@@ -165,6 +173,18 @@ func (a *FlowableActivity) SetupTableSchema(
 	})
 	defer shutdown()
 
+	// have to fetch config from the DB
+	cfg, err := internal.FetchConfigFromDB(config.FlowName)
+	if err != nil {
+		return a.Alerter.LogFlowError(ctx, config.FlowName, fmt.Errorf("failed to fetch config from DB: %w", err))
+	}
+	tableMappings := cfg.TableMappings
+	if len(config.FilteredTableMappings) > 0 {
+		// we use the filtered table mappings if provided. they are provided from
+		// the sync flow which includes changes to the schema.
+		tableMappings = config.FilteredTableMappings
+	}
+
 	logger := internal.LoggerFromCtx(ctx)
 	ctx = context.WithValue(ctx, shared.FlowNameKey, config.FlowName)
 	srcConn, err := connectors.GetByNameAs[connectors.GetTableSchemaConnector](ctx, config.Env, a.CatalogPool, config.PeerName)
@@ -173,11 +193,11 @@ func (a *FlowableActivity) SetupTableSchema(
 	}
 	defer connectors.CloseConnector(ctx, srcConn)
 
-	tableNameSchemaMapping, err := srcConn.GetTableSchema(ctx, config.Env, config.Version, config.System, config.TableMappings)
+	tableNameSchemaMapping, err := srcConn.GetTableSchema(ctx, config.Env, config.Version, config.System, tableMappings)
 	if err != nil {
 		return a.Alerter.LogFlowError(ctx, config.FlowName, fmt.Errorf("failed to get GetTableSchemaConnector: %w", err))
 	}
-	processed := internal.BuildProcessedSchemaMapping(config.TableMappings, tableNameSchemaMapping, logger)
+	processed := internal.BuildProcessedSchemaMapping(tableMappings, tableNameSchemaMapping, logger)
 
 	tx, err := a.CatalogPool.BeginTx(ctx, pgx.TxOptions{})
 	if err != nil {
@@ -213,6 +233,13 @@ func (a *FlowableActivity) CreateNormalizedTable(
 	numTablesSetup := atomic.Uint32{}
 	numTablesToSetup := atomic.Int32{}
 
+	cfg, err := internal.FetchConfigFromDB(config.FlowName)
+	if err != nil {
+		return nil, a.Alerter.LogFlowError(ctx, config.FlowName, fmt.Errorf("failed to fetch config from DB: %w", err))
+	}
+
+	tableMappings := cfg.TableMappings
+
 	shutdown := heartbeatRoutine(ctx, func() string {
 		return fmt.Sprintf("setting up normalized tables - %d of %d done", numTablesSetup.Load(), numTablesToSetup.Load())
 	})
@@ -244,7 +271,7 @@ func (a *FlowableActivity) CreateNormalizedTable(
 
 	numTablesToSetup.Store(int32(len(tableNameSchemaMapping)))
 	tableExistsMapping := make(map[string]bool, len(tableNameSchemaMapping))
-	for _, tableMapping := range config.TableMappings {
+	for _, tableMapping := range tableMappings {
 		tableIdentifier := tableMapping.DestinationTableIdentifier
 		tableSchema := tableNameSchemaMapping[tableIdentifier]
 		existing, err := conn.SetupNormalizedTable(
@@ -292,6 +319,12 @@ func (a *FlowableActivity) SyncFlow(
 	var syncingBatchID atomic.Int64
 	var syncState atomic.Pointer[string]
 	syncState.Store(shared.Ptr("setup"))
+
+	config, err := internal.FetchConfigFromDB(config.FlowJobName)
+	if err != nil {
+		return fmt.Errorf("unable to query flow config from catalog: %w", err)
+	}
+
 	shutdown := heartbeatRoutine(ctx, func() string {
 		// Must load Waiting after BatchID to avoid race saying we're waiting on currently processing batch
 		sBatchID := syncingBatchID.Load()
@@ -363,10 +396,10 @@ func (a *FlowableActivity) SyncFlow(
 		var syncResponse *model.SyncResponse
 		var syncErr error
 		if config.System == protos.TypeSystem_Q {
-			syncResponse, syncErr = a.syncRecords(groupCtx, config, options, srcConn.(connectors.CDCPullConnector),
+			syncResponse, syncErr = a.syncRecords(groupCtx, config, srcConn.(connectors.CDCPullConnector),
 				normRequests, normResponses, normBufferSize, &syncingBatchID, &syncState)
 		} else {
-			syncResponse, syncErr = a.syncPg(groupCtx, config, options, srcConn.(connectors.CDCPullPgConnector),
+			syncResponse, syncErr = a.syncPg(groupCtx, config, srcConn.(connectors.CDCPullPgConnector),
 				normRequests, normResponses, normBufferSize, &syncingBatchID, &syncState)
 		}
 
@@ -414,7 +447,6 @@ func (a *FlowableActivity) SyncFlow(
 func (a *FlowableActivity) syncRecords(
 	ctx context.Context,
 	config *protos.FlowConnectionConfigs,
-	options *protos.SyncFlowOptions,
 	srcConn connectors.CDCPullConnector,
 	normRequests *concurrency.LastChan,
 	normResponses *concurrency.LastChan,
@@ -451,7 +483,7 @@ func (a *FlowableActivity) syncRecords(
 			return stream, nil
 		}
 	}
-	return syncCore(ctx, a, config, options, srcConn,
+	return syncCore(ctx, a, config, srcConn,
 		normRequests, normResponses, normBufferSize,
 		syncingBatchID, syncWaiting, adaptStream,
 		connectors.CDCPullConnector.PullRecords,
@@ -461,15 +493,14 @@ func (a *FlowableActivity) syncRecords(
 func (a *FlowableActivity) syncPg(
 	ctx context.Context,
 	config *protos.FlowConnectionConfigs,
-	options *protos.SyncFlowOptions,
 	srcConn connectors.CDCPullPgConnector,
 	normRequests *concurrency.LastChan,
 	normResponses *concurrency.LastChan,
 	normBufferSize int64,
 	syncingBatchID *atomic.Int64,
 	syncWaiting *atomic.Pointer[string],
 ) (*model.SyncResponse, error) {
-	return syncCore(ctx, a, config, options, srcConn,
+	return syncCore(ctx, a, config, srcConn,
 		normRequests, normResponses, normBufferSize,
 		syncingBatchID, syncWaiting, nil,
 		connectors.CDCPullPgConnector.PullPg,
@@ -1214,9 +1245,13 @@ func (a *FlowableActivity) ReplicateXminPartition(ctx context.Context,
 	}
 }
 
-func (a *FlowableActivity) AddTablesToPublication(ctx context.Context, cfg *protos.FlowConnectionConfigs,
+func (a *FlowableActivity) AddTablesToPublication(ctx context.Context, flowJobName string,
 	additionalTableMappings []*protos.TableMapping,
 ) error {
+	cfg, err := internal.FetchConfigFromDB(flowJobName)
+	if err != nil {
+		return fmt.Errorf("unable to query flow config from catalog: %w", err)
+	}
 	ctx = context.WithValue(ctx, shared.FlowNameKey, cfg.FlowJobName)
 	srcConn, err := connectors.GetByNameAs[*connpostgres.PostgresConnector](ctx, cfg.Env, a.CatalogPool, cfg.SourceName)
 	if err != nil {

flow/activities/flowable_core.go

@@ -80,11 +80,10 @@ func (a *FlowableActivity) getTableNameSchemaMapping(ctx context.Context, flowNa
 func (a *FlowableActivity) applySchemaDeltas(
 	ctx context.Context,
 	config *protos.FlowConnectionConfigs,
-	options *protos.SyncFlowOptions,
 	schemaDeltas []*protos.TableSchemaDelta,
 ) error {
 	filteredTableMappings := make([]*protos.TableMapping, 0, len(schemaDeltas))
-	for _, tableMapping := range options.TableMappings {
+	for _, tableMapping := range config.TableMappings {
 		if slices.ContainsFunc(schemaDeltas, func(schemaDelta *protos.TableSchemaDelta) bool {
 			return schemaDelta.SrcTableName == tableMapping.SourceTableIdentifier &&
 				schemaDelta.DstTableName == tableMapping.DestinationTableIdentifier
@@ -95,12 +94,12 @@ func (a *FlowableActivity) applySchemaDeltas(
 
 	if len(schemaDeltas) > 0 {
 		if err := a.SetupTableSchema(ctx, &protos.SetupTableSchemaBatchInput{
-			PeerName:      config.SourceName,
-			TableMappings: filteredTableMappings,
-			FlowName:      config.FlowJobName,
-			System:        config.System,
-			Env:           config.Env,
-			Version:       config.Version,
+			PeerName:              config.SourceName,
+			FilteredTableMappings: filteredTableMappings,
+			FlowName:              config.FlowJobName,
+			System:                config.System,
+			Env:                   config.Env,
+			Version:               config.Version,
 		}); err != nil {
 			return a.Alerter.LogFlowError(ctx, config.FlowJobName, fmt.Errorf("failed to execute schema update at source: %w", err))
 		}
@@ -112,7 +111,6 @@ func syncCore[TPull connectors.CDCPullConnectorCore, TSync connectors.CDCSyncCon
 	ctx context.Context,
 	a *FlowableActivity,
 	config *protos.FlowConnectionConfigs,
-	options *protos.SyncFlowOptions,
 	srcConn TPull,
 	normRequests *concurrency.LastChan,
 	normResponses *concurrency.LastChan,
@@ -127,16 +125,17 @@ func syncCore[TPull connectors.CDCPullConnectorCore, TSync connectors.CDCSyncCon
 	ctx = context.WithValue(ctx, shared.FlowNameKey, flowName)
 	logger := internal.LoggerFromCtx(ctx)
 
-	tblNameMapping := make(map[string]model.NameAndExclude, len(options.TableMappings))
-	for _, v := range options.TableMappings {
+	// we should be able to rely on `config` here.
+	tblNameMapping := make(map[string]model.NameAndExclude, len(config.TableMappings))
+	for _, v := range config.TableMappings {
 		tblNameMapping[v.SourceTableIdentifier] = model.NewNameAndExclude(v.DestinationTableIdentifier, v.Exclude)
 	}
 
 	if err := srcConn.ConnectionActive(ctx); err != nil {
 		return nil, temporal.NewNonRetryableApplicationError("connection to source down", "disconnect", nil)
 	}
 
-	batchSize := options.BatchSize
+	batchSize := config.MaxBatchSize
 	if batchSize == 0 {
 		batchSize = 250_000
 	}
@@ -189,13 +188,13 @@ func syncCore[TPull connectors.CDCPullConnectorCore, TSync connectors.CDCSyncCon
 	errGroup.Go(func() error {
 		return pull(srcConn, errCtx, a.CatalogPool, a.OtelManager, &model.PullRecordsRequest[Items]{
 			FlowJobName:           flowName,
-			SrcTableIDNameMapping: options.SrcTableIdNameMapping,
+			SrcTableIDNameMapping: config.SrcTableIdNameMapping,
 			TableNameMapping:      tblNameMapping,
 			LastOffset:            lastOffset,
 			ConsumedOffset:        &consumedOffset,
 			MaxBatchSize:          batchSize,
 			IdleTimeout: internal.PeerDBCDCIdleTimeoutSeconds(
-				int(options.IdleTimeoutSeconds),
+				int(config.IdleTimeoutSeconds),
 			),
 			TableNameSchemaMapping:      tableNameSchemaMapping,
 			OverridePublicationName:     config.PublicationName,
@@ -231,11 +230,11 @@ func syncCore[TPull connectors.CDCPullConnectorCore, TSync connectors.CDCSyncCon
 		defer connectors.CloseConnector(ctx, dstConn)
 
 		syncState.Store(shared.Ptr("updating schema"))
-		if err := dstConn.ReplayTableSchemaDeltas(ctx, config.Env, flowName, options.TableMappings, recordBatchSync.SchemaDeltas); err != nil {
+		if err := dstConn.ReplayTableSchemaDeltas(ctx, config.Env, flowName, config.TableMappings, recordBatchSync.SchemaDeltas); err != nil {
 			return nil, fmt.Errorf("failed to sync schema: %w", err)
 		}
 
-		return nil, a.applySchemaDeltas(ctx, config, options, recordBatchSync.SchemaDeltas)
+		return nil, a.applySchemaDeltas(ctx, config, recordBatchSync.SchemaDeltas)
 	}
 
 	var res *model.SyncResponse
@@ -268,7 +267,7 @@ func syncCore[TPull connectors.CDCPullConnectorCore, TSync connectors.CDCSyncCon
 			Records:                recordBatchSync,
 			ConsumedOffset:         &consumedOffset,
 			FlowJobName:            flowName,
-			TableMappings:          options.TableMappings,
+			TableMappings:          config.TableMappings,
 			StagingPath:            config.CdcStagingPath,
 			Script:                 config.Script,
 			TableNameSchemaMapping: tableNameSchemaMapping,
@@ -331,7 +330,7 @@ func syncCore[TPull connectors.CDCPullConnectorCore, TSync connectors.CDCSyncCon
 	a.OtelManager.Metrics.CurrentBatchIdGauge.Record(ctx, res.CurrentSyncBatchID)
 
 	syncState.Store(shared.Ptr("updating schema"))
-	if err := a.applySchemaDeltas(ctx, config, options, res.TableSchemaDeltas); err != nil {
+	if err := a.applySchemaDeltas(ctx, config, res.TableSchemaDeltas); err != nil {
 		return nil, err
 	}
 

flow/cmd/handler.go

@@ -75,7 +75,6 @@ func (h *FlowRequestHandler) createCdcJobEntry(ctx context.Context,
 	if err != nil {
 		return fmt.Errorf("unable to marshal flow config: %w", err)
 	}
-
 	if _, err := h.pool.Exec(ctx,
 		`INSERT INTO flows (workflow_id, name, source_peer, destination_peer, config_proto, status,
 		description, source_table_identifier, destination_table_identifier) VALUES ($1,$2,$3,$4,$5,$6,'gRPC','','')`,
@@ -155,7 +154,7 @@ func (h *FlowRequestHandler) CreateCDCFlow(
 		return nil, fmt.Errorf("unable to create flow job entry: %w", err)
 	}
 
-	if _, err := h.temporalClient.ExecuteWorkflow(ctx, workflowOptions, peerflow.CDCFlowWorkflow, cfg, nil); err != nil {
+	if _, err := h.temporalClient.ExecuteWorkflow(ctx, workflowOptions, peerflow.CDCFlowWorkflow, cfg.FlowJobName, nil); err != nil {
 		slog.Error("unable to start PeerFlow workflow", slog.Any("error", err))
 		return nil, fmt.Errorf("unable to start PeerFlow workflow: %w", err)
 	}

flow/cmd/mirror_status.go

@@ -18,7 +18,6 @@ import (
 	"github.com/PeerDB-io/peerdb/flow/generated/protos"
 	"github.com/PeerDB-io/peerdb/flow/internal"
 	"github.com/PeerDB-io/peerdb/flow/shared"
-	peerflow "github.com/PeerDB-io/peerdb/flow/workflows"
 )
 
 func (h *FlowRequestHandler) ListMirrors(
@@ -137,23 +136,6 @@ func (h *FlowRequestHandler) cdcFlowStatus(
 		slog.Error("unable to query flow config from catalog", slog.Any("error", err))
 		return nil, err
 	}
-	workflowID, err := h.getWorkflowID(ctx, req.FlowJobName)
-	if err != nil {
-		slog.Error("unable to get the workflow ID of mirror", slog.Any("error", err))
-		return nil, err
-	}
-	state, err := h.getCDCWorkflowState(ctx, workflowID)
-	if err != nil {
-		slog.Error("unable to get the state of mirror", slog.Any("error", err))
-		return nil, err
-	}
-
-	// patching config to show latest values from state
-	if state.SyncFlowOptions != nil {
-		config.IdleTimeoutSeconds = state.SyncFlowOptions.IdleTimeoutSeconds
-		config.MaxBatchSize = state.SyncFlowOptions.BatchSize
-		config.TableMappings = state.SyncFlowOptions.TableMappings
-	}
 
 	srcType, err := connectors.LoadPeerType(ctx, h.pool, config.SourceName)
 	if err != nil {
@@ -462,7 +444,7 @@ func (h *FlowRequestHandler) getFlowConfigFromCatalog(
 ) (*protos.FlowConnectionConfigs, error) {
 	var configBytes sql.RawBytes
 	if err := h.pool.QueryRow(ctx,
-		"SELECT config_proto FROM flows WHERE name = $1", flowJobName,
+		"SELECT config_proto FROM flows WHERE name = $1 LIMIT 1", flowJobName,
 	).Scan(&configBytes); err != nil {
 		slog.Error("unable to query flow config from catalog", slog.Any("error", err))
 		return nil, fmt.Errorf("unable to query flow config from catalog: %w", err)
@@ -492,24 +474,6 @@ func (h *FlowRequestHandler) getWorkflowStatus(ctx context.Context, workflowID s
 	return internal.GetWorkflowStatus(ctx, h.pool, workflowID)
 }
 
-func (h *FlowRequestHandler) getCDCWorkflowState(ctx context.Context,
-	workflowID string,
-) (*peerflow.CDCFlowWorkflowState, error) {
-	res, err := h.temporalClient.QueryWorkflow(ctx, workflowID, "", shared.CDCFlowStateQuery)
-	if err != nil {
-		slog.Error(fmt.Sprintf("failed to get state in workflow with ID %s: %s", workflowID, err.Error()))
-		return nil,
-			fmt.Errorf("failed to get state in workflow with ID %s: %w", workflowID, err)
-	}
-	var state peerflow.CDCFlowWorkflowState
-	if err := res.Get(&state); err != nil {
-		slog.Error(fmt.Sprintf("failed to get state in workflow with ID %s: %s", workflowID, err.Error()))
-		return nil,
-			fmt.Errorf("failed to get state in workflow with ID %s: %w", workflowID, err)
-	}
-	return &state, nil
-}
-
 func (h *FlowRequestHandler) getMirrorCreatedAt(ctx context.Context, flowJobName string) (*time.Time, error) {
 	var createdAt pgtype.Timestamp
 	err := h.pool.QueryRow(ctx, "SELECT created_at FROM flows WHERE name=$1", flowJobName).Scan(&createdAt)

flow/connectors/clickhouse/normalize.go

@@ -89,7 +89,12 @@ func (c *ClickHouseConnector) generateCreateTableSQLForNormalizedTable(
 	tmEngine := protos.TableEngine_CH_ENGINE_REPLACING_MERGE_TREE
 
 	var tableMapping *protos.TableMapping
-	for _, tm := range config.TableMappings {
+
+	cfg, err := internal.FetchConfigFromDB(config.FlowName)
+	if err != nil {
+		return nil, fmt.Errorf("failed to fetch config from DB: %w", err)
+	}
+	for _, tm := range cfg.TableMappings {
 		if tm.DestinationTableIdentifier == tableIdentifier {
 			tmEngine = tm.Engine
 			tableMapping = tm