rebase & addr feedback

Author: alexstoick

flow/activities/flowable.go

@@ -1077,13 +1077,12 @@ func (a *FlowableActivity) emitLogRetentionHours(
 }
 
 var activeFlowStatuses = map[protos.FlowStatus]struct{}{
-	protos.FlowStatus_STATUS_RUNNING:   {},
-	protos.FlowStatus_STATUS_PAUSED:    {},
-	protos.FlowStatus_STATUS_PAUSING:   {},
-	protos.FlowStatus_STATUS_SETUP:     {},
-	protos.FlowStatus_STATUS_SNAPSHOT:  {},
-	protos.FlowStatus_STATUS_RESYNC:    {},
-	protos.FlowStatus_STATUS_MODIFYING: {},
+	protos.FlowStatus_STATUS_RUNNING:  {},
+	protos.FlowStatus_STATUS_PAUSED:   {},
+	protos.FlowStatus_STATUS_PAUSING:  {},
+	protos.FlowStatus_STATUS_SETUP:    {},
+	protos.FlowStatus_STATUS_SNAPSHOT: {},
+	protos.FlowStatus_STATUS_RESYNC:   {},
 }
 
 func (a *FlowableActivity) QRepHasNewRows(ctx context.Context,

flow/activities/maintenance_activity.go

@@ -94,12 +94,11 @@ func (a *MaintenanceActivity) WaitForRunningSnapshotsAndIntermediateStates(
 }
 
 var waitStatuses map[protos.FlowStatus]struct{} = map[protos.FlowStatus]struct{}{
-	protos.FlowStatus_STATUS_SNAPSHOT:  {},
-	protos.FlowStatus_STATUS_SETUP:     {},
-	protos.FlowStatus_STATUS_RESYNC:    {},
-	protos.FlowStatus_STATUS_UNKNOWN:   {},
-	protos.FlowStatus_STATUS_PAUSING:   {},
-	protos.FlowStatus_STATUS_MODIFYING: {},
+	protos.FlowStatus_STATUS_SNAPSHOT: {},
+	protos.FlowStatus_STATUS_SETUP:    {},
+	protos.FlowStatus_STATUS_RESYNC:   {},
+	protos.FlowStatus_STATUS_UNKNOWN:  {},
+	protos.FlowStatus_STATUS_PAUSING:  {},
 }
 
 func (a *MaintenanceActivity) checkAndWaitIfNeeded(

flow/cmd/handler.go

@@ -158,7 +158,7 @@ func (h *FlowRequestHandler) CreateCDCFlow(
 		return nil, exceptions.NewInternalApiError(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, internal.CreateMinimalConfigFromFlowJobName(cfg.FlowJobName), nil); err != nil {
 		slog.ErrorContext(ctx, "unable to start PeerFlow workflow", slog.Any("error", err))
 		return nil, exceptions.NewInternalApiError(fmt.Errorf("unable to start PeerFlow workflow: %w", err))
 	}
@@ -428,7 +428,8 @@ func (h *FlowRequestHandler) CreatePeer(
 	ctx context.Context,
 	req *protos.CreatePeerRequest,
 ) (*protos.CreatePeerResponse, error) {
-	if !req.DisableValidation {
+	// Always validate peers
+	if true {
 		status, validateErr := h.ValidatePeer(ctx, &protos.ValidatePeerRequest{Peer: req.Peer})
 		if validateErr != nil {
 			// ValidatePeer returns proper grpc errors

flow/cmd/mirror_status.go

@@ -148,23 +148,7 @@ func (h *FlowRequestHandler) cdcFlowStatus(
 		slog.ErrorContext(ctx, "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.ErrorContext(ctx, "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.ErrorContext(ctx, "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
-	}
+	// The config is now always sourced from DB, so no state patching needed
 
 	srcType, err := connectors.LoadPeerType(ctx, h.pool, config.SourceName)
 	if err != nil {

flow/internal/flow_configuration_helpers.go

@@ -36,3 +36,11 @@ func FetchConfigFromDB(flowName string) (*protos.FlowConnectionConfigs, error) {
 
 	return &cfgFromDB, nil
 }
+
+// CreateMinimalConfigFromFlowJobName creates a minimal FlowConnectionConfigs with just the FlowJobName set.
+// This is used when calling CDCFlowWorkflow which will fetch the full config from DB internally.
+func CreateMinimalConfigFromFlowJobName(flowJobName string) *protos.FlowConnectionConfigs {
+	return &protos.FlowConnectionConfigs{
+		FlowJobName: flowJobName,
+	}
+}

flow/workflows/cdc_flow.go

@@ -240,21 +240,22 @@ func processTableAdditions(
 			additionalTablesUUID := GetUUID(ctx)
 			childAdditionalTablesCDCFlowID := GetChildWorkflowID("additional-cdc-flow", cfg.FlowJobName, additionalTablesUUID)
 			additionalTablesCfg := proto.CloneOf(cfg)
-			additionalTablesCfg.DoInitialSnapshot = !flowConfigUpdate.SkipInitialSnapshotForTableAdditions
+			// Default to doing initial snapshot for additional tables
+			additionalTablesCfg.DoInitialSnapshot = true
 			additionalTablesCfg.InitialSnapshotOnly = true
 			additionalTablesCfg.TableMappings = append(additionalTablesCfg.TableMappings, flowConfigUpdate.AdditionalTables...)
 			additionalTablesCfg.Resync = false
-			if state.SnapshotNumRowsPerPartition > 0 {
-				additionalTablesCfg.SnapshotNumRowsPerPartition = state.SnapshotNumRowsPerPartition
+			if flowConfigUpdate.SnapshotNumRowsPerPartition > 0 {
+				additionalTablesCfg.SnapshotNumRowsPerPartition = flowConfigUpdate.SnapshotNumRowsPerPartition
 			}
-			if state.SnapshotNumPartitionsOverride > 0 {
-				additionalTablesCfg.SnapshotNumPartitionsOverride = state.SnapshotNumPartitionsOverride
+			if flowConfigUpdate.SnapshotNumPartitionsOverride > 0 {
+				additionalTablesCfg.SnapshotNumPartitionsOverride = flowConfigUpdate.SnapshotNumPartitionsOverride
 			}
-			if state.SnapshotMaxParallelWorkers > 0 {
-				additionalTablesCfg.SnapshotMaxParallelWorkers = state.SnapshotMaxParallelWorkers
+			if flowConfigUpdate.SnapshotMaxParallelWorkers > 0 {
+				additionalTablesCfg.SnapshotMaxParallelWorkers = flowConfigUpdate.SnapshotMaxParallelWorkers
 			}
-			if state.SnapshotNumTablesInParallel > 0 {
-				additionalTablesCfg.SnapshotNumTablesInParallel = state.SnapshotNumTablesInParallel
+			if flowConfigUpdate.SnapshotNumTablesInParallel > 0 {
+				additionalTablesCfg.SnapshotNumTablesInParallel = flowConfigUpdate.SnapshotNumTablesInParallel
 			}
 
 			uploadConfigToCatalog(ctx, additionalTablesCfg)
@@ -274,7 +275,7 @@ func processTableAdditions(
 			childAddTablesCDCFlowFuture := workflow.ExecuteChildWorkflow(
 				childAddTablesCDCFlowCtx,
 				CDCFlowWorkflow,
-				additionalTablesCfg.FlowJobName,
+				internal.CreateMinimalConfigFromFlowJobName(additionalTablesCfg.FlowJobName),
 				nil, // nil is passed to trigger `setup` flow.
 			)
 			addTablesSelector.AddFuture(childAddTablesCDCFlowFuture, func(f workflow.Future) {
@@ -283,15 +284,10 @@ func processTableAdditions(
 		}
 	})
 
-	// additional tables should also be resynced, we don't know how much was done so far
-	// state.SyncFlowOptions.TableMappings = append(state.SyncFlowOptions.TableMappings, flowConfigUpdate.AdditionalTables...)
-
 	for res == nil {
 		addTablesSelector.Select(ctx)
 		if state.ActiveSignal == model.TerminateSignal || state.ActiveSignal == model.ResyncSignal {
 			if state.ActiveSignal == model.ResyncSignal {
-				// additional tables should also be resynced, we don't know how much was done so far
-				// state.SyncFlowOptions.TableMappings = append(state.SyncFlowOptions.TableMappings, flowConfigUpdate.AdditionalTables...)
 				resyncCfg := syncStateToConfigProtoInCatalog(ctx, cfg, state.FlowConfigUpdate)
 				state.DropFlowInput.FlowJobName = resyncCfg.FlowJobName
 				state.DropFlowInput.FlowConnectionConfigs = resyncCfg
@@ -318,7 +314,7 @@ func processTableRemovals(
 	cfg *protos.FlowConnectionConfigs,
 	state *CDCFlowWorkflowState,
 ) error {
-	state.updateStatus(ctx, logger, protos.FlowStatus_STATUS_MODIFYING)
+	state.updateStatus(ctx, logger, protos.FlowStatus_STATUS_RUNNING)
 	removeTablesSelector := workflow.NewNamedSelector(ctx, "RemoveTables")
 	removeTablesSelector.AddReceive(ctx.Done(), func(_ workflow.ReceiveChannel, _ bool) {})
 	flowSignalStateChangeChan := model.FlowSignalStateChange.GetSignalChannel(ctx)
@@ -428,24 +424,25 @@ func addCdcPropertiesSignalListener(
 			slog.Uint64("SnapshotNumPartitionsOverride", uint64(cdcConfigUpdate.SnapshotNumPartitionsOverride)),
 			slog.Uint64("SnapshotMaxParallelWorkers", uint64(cdcConfigUpdate.SnapshotMaxParallelWorkers)),
 			slog.Uint64("SnapshotNumTablesInParallel", uint64(cdcConfigUpdate.SnapshotNumTablesInParallel)),
-			slog.Bool("SkipInitialSnapshotForTableAdditions", cdcConfigUpdate.SkipInitialSnapshotForTableAdditions),
 		)
 	})
 }
 
 func CDCFlowWorkflow(
 	ctx workflow.Context,
-	flowJobName string,
-	// cfg *protos.FlowConnectionConfigs,
+	cfg *protos.FlowConnectionConfigs,
 	state *CDCFlowWorkflowState,
 ) (*CDCFlowWorkflowResult, error) {
-	cfg, err := internal.FetchConfigFromDB(flowJobName)
-	if cfg == nil {
-		return nil, errors.New("invalid connection configs")
-	}
+	// Fetch full config from DB using the flowJobName from the minimal config
+	cfgFromDB, err := internal.FetchConfigFromDB(cfg.FlowJobName)
 	if err != nil {
-		return nil, fmt.Errorf("unable to unmarshal flow config: %w", err)
+		return nil, fmt.Errorf("unable to fetch flow config from DB: %w", err)
+	}
+	if cfgFromDB == nil {
+		return nil, errors.New("invalid connection configs")
 	}
+	// Use the config from DB for all operations
+	cfg = cfgFromDB
 
 	logger := log.With(workflow.GetLogger(ctx), slog.String(string(shared.FlowNameKey), cfg.FlowJobName))
 	if state == nil {
@@ -533,7 +530,7 @@ func CDCFlowWorkflow(
 
 		logger.Info("mirror resumed", slog.Duration("after", time.Since(startTime)))
 		state.updateStatus(ctx, logger, protos.FlowStatus_STATUS_RUNNING)
-		return state, workflow.NewContinueAsNewError(ctx, CDCFlowWorkflow, cfg.FlowJobName, state)
+		return state, workflow.NewContinueAsNewError(ctx, CDCFlowWorkflow, internal.CreateMinimalConfigFromFlowJobName(cfg.FlowJobName), state)
 	}
 
 	originalRunID := workflow.GetInfo(ctx).OriginalRunID
@@ -678,7 +675,7 @@ func CDCFlowWorkflow(
 			}
 		}
 
-		// TODOAS: here we will also store the table mappings in the state.
+		// here we will also store the table mappings in the state.
 		maps.Copy(cfg.SrcTableIdNameMapping, setupFlowOutput.SrcTableIdNameMapping)
 		uploadConfigToCatalog(ctx, cfg)
 
@@ -701,7 +698,7 @@ func CDCFlowWorkflow(
 		// during any operation that triggers another snapshot (INCLUDING add tables).
 		// this could fail for very weird Temporal resets
 
-		// TODOAS : this will send the additionalTables to `temporal`, meaning
+		// This will send the additionalTables to `temporal`, meaning
 		// that we cannot add too many tables at once, or we risk the blob is too
 		// large (2MB limit).
 		snapshotFlowFuture := workflow.ExecuteChildWorkflow(
@@ -803,7 +800,7 @@ func CDCFlowWorkflow(
 			logger.Info("executed setup flow and snapshot flow, start running")
 			state.updateStatus(ctx, logger, protos.FlowStatus_STATUS_RUNNING)
 		}
-		return state, workflow.NewContinueAsNewError(ctx, CDCFlowWorkflow, cfg.FlowJobName, state)
+		return state, workflow.NewContinueAsNewError(ctx, CDCFlowWorkflow, internal.CreateMinimalConfigFromFlowJobName(cfg.FlowJobName), state)
 	}
 
 	var finished bool
@@ -933,7 +930,7 @@ func CDCFlowWorkflow(
 			if state.ActiveSignal == model.TerminateSignal || state.ActiveSignal == model.ResyncSignal {
 				return state, workflow.NewContinueAsNewError(ctx, DropFlowWorkflow, state.DropFlowInput)
 			}
-			return state, workflow.NewContinueAsNewError(ctx, CDCFlowWorkflow, cfg.FlowJobName, state)
+			return state, workflow.NewContinueAsNewError(ctx, CDCFlowWorkflow, internal.CreateMinimalConfigFromFlowJobName(cfg.FlowJobName), state)
 		}
 	}
 }

flow/workflows/drop_flow.go

@@ -11,6 +11,7 @@ import (
 	"go.temporal.io/sdk/workflow"
 
 	"github.com/PeerDB-io/peerdb/flow/generated/protos"
+	"github.com/PeerDB-io/peerdb/flow/internal"
 	"github.com/PeerDB-io/peerdb/flow/model"
 	"github.com/PeerDB-io/peerdb/flow/shared"
 )
@@ -211,7 +212,7 @@ func DropFlowWorkflow(ctx workflow.Context, input *protos.DropFlowInput) error {
 	}
 
 	if input.Resync {
-		return workflow.NewContinueAsNewError(ctx, CDCFlowWorkflow, input.FlowConnectionConfigs, nil)
+		return workflow.NewContinueAsNewError(ctx, CDCFlowWorkflow, internal.CreateMinimalConfigFromFlowJobName(input.FlowConnectionConfigs.FlowJobName), nil)
 	}
 
 	return nil