Replace local activities with plain activities, except status in catalog (#3350)

Like #3301 but trying to take the status in catalog separately

> this small optimization increases complexity around reexecution
> moving towards keeping our use of temporal api surface minimal
> if it's found that a workflow is spending too much time dealing with overhead of activities,
> they should be combined into one long running activity

---------

Co-authored-by: Philip Dubé <[email protected]>

Author: ilidemi

flow/activities/flowable.go

@@ -1412,3 +1412,11 @@ func (a *FlowableActivity) GetFlowMetadata(
 		IsResync: input.IsResync,
 	}, nil
 }
+
+func (a *FlowableActivity) UpdateCDCConfigInCatalogActivity(ctx context.Context, cfg *protos.FlowConnectionConfigs) error {
+	return internal.UpdateCDCConfigInCatalog(ctx, a.CatalogPool, internal.LoggerFromCtx(ctx), cfg)
+}
+
+func (a *FlowableActivity) PeerDBFullRefreshOverwriteMode(ctx context.Context, env map[string]string) (bool, error) {
+	return internal.PeerDBFullRefreshOverwriteMode(ctx, env)
+}

flow/activities/snapshot_activity.go

@@ -167,6 +167,10 @@ func (a *SnapshotActivity) LoadTableSchema(
 	return internal.LoadTableSchemaFromCatalog(ctx, a.CatalogPool, flowName, tableName)
 }
 
+func (a *SnapshotActivity) GetPeerType(ctx context.Context, name string) (protos.DBType, error) {
+	return connectors.LoadPeerType(ctx, a.CatalogPool, name)
+}
+
 func (a *SnapshotActivity) GetDefaultPartitionKeyForTables(
 	ctx context.Context,
 	input *protos.FlowConnectionConfigs,

flow/e2e/clickhouse/peer_flow_ch_test.go

@@ -147,8 +147,9 @@ func (s ClickHouseSuite) Test_Addition_Removal() {
 	e2e.EnvWaitFor(s.t, env, 4*time.Minute, "adding table", func() bool {
 		return env.GetFlowStatus(s.t) == protos.FlowStatus_STATUS_RUNNING
 	})
-	afterAddRunID := e2e.EnvGetRunID(s.t, env)
-	require.NotEqual(s.t, runID, afterAddRunID)
+	e2e.EnvWaitFor(s.t, env, time.Minute, "ContinueAsNew", func() bool {
+		return runID != e2e.EnvGetRunID(s.t, env)
+	})
 
 	require.NoError(s.t, s.source.Exec(s.t.Context(), fmt.Sprintf(`INSERT INTO %s ("key") VALUES ('test')`, addedSrcTableName)))
 	e2e.EnvWaitForEqualTablesWithNames(env, s, "first insert to added table", "test_table_add_remove_added", addedDstTableName, "id,\"key\"")
@@ -497,11 +498,10 @@ func (s ClickHouseSuite) WeirdTable(tableName string) {
 	_, err = s.Conn().Exec(s.t.Context(),
 		fmt.Sprintf("INSERT INTO %s (key, \"includedColumn?\", \"excludedColumn?\") VALUES ('cdc','still','ex')", srcFullName))
 	require.NoError(s.t, err)
-
 	e2e.EnvWaitForEqualTablesWithNames(env, s, "waiting on cdc", srcTableName, dstTableName, "id,\"key\"")
-
 	env.Cancel(s.t.Context())
 	e2e.RequireEnvCanceled(s.t, env)
+
 	env = e2e.ExecuteWorkflow(s.t.Context(), tc, shared.PeerFlowTaskQueue, peerflow.DropFlowWorkflow, &protos.DropFlowInput{
 		FlowJobName:           flowConnConfig.FlowJobName,
 		DropFlowStats:         false,

flow/workflows/cdc_flow.go

@@ -72,8 +72,8 @@ func syncStatusToCatalog(ctx workflow.Context, logger log.Logger, status protos.
 		StartToCloseTimeout: 1 * time.Minute,
 	})
 
-	updateFuture := workflow.ExecuteLocalActivity(updateCtx, updateFlowStatusInCatalogActivity,
-		workflow.GetInfo(ctx).WorkflowExecution.ID, status)
+	updateFuture := workflow.ExecuteLocalActivity(updateCtx,
+		updateFlowStatusInCatalogActivity, workflow.GetInfo(ctx).WorkflowExecution.ID, status)
 	if err := updateFuture.Get(updateCtx, nil); err != nil {
 		logger.Error("Failed to update flow status in catalog", slog.Any("error", err), slog.String("flowStatus", status.String()))
 	}
@@ -130,14 +130,13 @@ func uploadConfigToCatalog(
 	ctx workflow.Context,
 	cfg *protos.FlowConnectionConfigs,
 ) {
-	updateCtx := workflow.WithLocalActivityOptions(ctx, workflow.LocalActivityOptions{
+	updateCtx := workflow.WithActivityOptions(ctx, workflow.ActivityOptions{
 		StartToCloseTimeout: 5 * time.Minute,
 	})
 
-	logger := workflow.GetLogger(ctx)
-	updateFuture := workflow.ExecuteLocalActivity(updateCtx, updateCDCConfigInCatalogActivity, logger, cfg)
+	updateFuture := workflow.ExecuteActivity(updateCtx, flowable.UpdateCDCConfigInCatalogActivity, cfg)
 	if err := updateFuture.Get(updateCtx, nil); err != nil {
-		logger.Warn("Failed to update CDC config in catalog", slog.Any("error", err))
+		workflow.GetLogger(ctx).Warn("Failed to update CDC config in catalog", slog.Any("error", err))
 	}
 }
 
@@ -178,24 +177,21 @@ func processCDCFlowConfigUpdate(
 
 	tablesAreAdded := len(flowConfigUpdate.AdditionalTables) > 0
 	tablesAreRemoved := len(flowConfigUpdate.RemovedTables) > 0
-	if !tablesAreAdded && !tablesAreRemoved {
-		syncStateToConfigProtoInCatalog(ctx, cfg, state)
-		return nil
-	}
-
-	logger.Info("processing CDCFlowConfigUpdate", slog.Any("updatedState", flowConfigUpdate))
+	if tablesAreAdded || tablesAreRemoved {
+		logger.Info("processing CDCFlowConfigUpdate", slog.Any("updatedState", flowConfigUpdate))
 
-	if tablesAreAdded {
-		if err := processTableAdditions(ctx, logger, cfg, state, mirrorNameSearch); err != nil {
-			logger.Error("failed to process additional tables", slog.Any("error", err))
-			return err
+		if tablesAreAdded {
+			if err := processTableAdditions(ctx, logger, cfg, state, mirrorNameSearch); err != nil {
+				logger.Error("failed to process additional tables", slog.Any("error", err))
+				return err
+			}
 		}
-	}
 
-	if tablesAreRemoved {
-		if err := processTableRemovals(ctx, logger, cfg, state); err != nil {
-			logger.Error("failed to process removed tables", slog.Any("error", err))
-			return err
+		if tablesAreRemoved {
+			if err := processTableRemovals(ctx, logger, cfg, state); err != nil {
+				logger.Error("failed to process removed tables", slog.Any("error", err))
+				return err
+			}
 		}
 	}
 
@@ -560,7 +556,7 @@ func CDCFlowWorkflow(
 			}
 		}
 
-		logger.Info(fmt.Sprintf("mirror has been resumed after %s", time.Since(startTime).Round(time.Second)))
+		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, state)
 	}

flow/workflows/local_activities.go

@@ -3,58 +3,11 @@ package peerflow
 import (
 	"context"
 	"fmt"
-	"log/slog"
-	"time"
 
-	"go.temporal.io/sdk/log"
-	"go.temporal.io/sdk/workflow"
-
-	"github.com/PeerDB-io/peerdb/flow/connectors"
 	"github.com/PeerDB-io/peerdb/flow/generated/protos"
 	"github.com/PeerDB-io/peerdb/flow/internal"
 )
 
-func getQRepOverwriteFullRefreshMode(wCtx workflow.Context, logger log.Logger, env map[string]string) bool {
-	checkCtx := workflow.WithLocalActivityOptions(wCtx, workflow.LocalActivityOptions{
-		StartToCloseTimeout: time.Minute,
-	})
-
-	getFullRefreshFuture := workflow.ExecuteLocalActivity(checkCtx, internal.PeerDBFullRefreshOverwriteMode, env)
-	var fullRefreshEnabled bool
-	if err := getFullRefreshFuture.Get(checkCtx, &fullRefreshEnabled); err != nil {
-		logger.Warn("Failed to check if full refresh mode is enabled", slog.Any("error", err))
-		return false
-	}
-	return fullRefreshEnabled
-}
-
-func localPeerType(ctx context.Context, name string) (protos.DBType, error) {
-	pool, err := internal.GetCatalogConnectionPoolFromEnv(ctx)
-	if err != nil {
-		return 0, err
-	}
-	return connectors.LoadPeerType(ctx, pool, name)
-}
-
-func getPeerType(wCtx workflow.Context, name string) (protos.DBType, error) {
-	checkCtx := workflow.WithLocalActivityOptions(wCtx, workflow.LocalActivityOptions{
-		StartToCloseTimeout: time.Minute,
-	})
-
-	getFuture := workflow.ExecuteLocalActivity(checkCtx, localPeerType, name)
-	var dbtype protos.DBType
-	err := getFuture.Get(checkCtx, &dbtype)
-	return dbtype, err
-}
-
-func updateCDCConfigInCatalogActivity(ctx context.Context, logger log.Logger, cfg *protos.FlowConnectionConfigs) error {
-	pool, err := internal.GetCatalogConnectionPoolFromEnv(ctx)
-	if err != nil {
-		return fmt.Errorf("failed to get catalog connection pool: %w", err)
-	}
-	return internal.UpdateCDCConfigInCatalog(ctx, pool, logger, cfg)
-}
-
 func updateFlowStatusInCatalogActivity(
 	ctx context.Context,
 	workflowID string,

flow/workflows/setup_flow.go

@@ -56,12 +56,12 @@ func (s *SetupFlowExecution) checkConnectionsAndSetupMetadataTables(
 ) error {
 	s.Info("checking connections for CDC flow")
 
-	checkCtx := workflow.WithLocalActivityOptions(ctx, workflow.LocalActivityOptions{
+	checkCtx := workflow.WithActivityOptions(ctx, workflow.ActivityOptions{
 		StartToCloseTimeout: time.Minute,
 	})
 
 	// first check the source peer connection
-	srcConnStatusFuture := workflow.ExecuteLocalActivity(checkCtx, flowable.CheckConnection, &protos.SetupInput{
+	srcConnStatusFuture := workflow.ExecuteActivity(checkCtx, flowable.CheckConnection, &protos.SetupInput{
 		Env:      config.Env,
 		PeerName: config.SourceName,
 		FlowName: config.FlowJobName,
@@ -71,7 +71,7 @@ func (s *SetupFlowExecution) checkConnectionsAndSetupMetadataTables(
 		PeerName: config.DestinationName,
 		FlowName: config.FlowJobName,
 	}
-	destConnStatusFuture := workflow.ExecuteLocalActivity(checkCtx, flowable.CheckMetadataTables, dstSetupInput)
+	destConnStatusFuture := workflow.ExecuteActivity(checkCtx, flowable.CheckMetadataTables, dstSetupInput)
 	if err := srcConnStatusFuture.Get(checkCtx, nil); err != nil {
 		return fmt.Errorf("failed to check source peer connection: %w", err)
 	}

flow/workflows/snapshot_flow.go

@@ -31,6 +31,16 @@ type SnapshotFlowExecution struct {
 	logger log.Logger
 }
 
+func getPeerType(wCtx workflow.Context, name string) (protos.DBType, error) {
+	checkCtx := workflow.WithActivityOptions(wCtx, workflow.ActivityOptions{
+		StartToCloseTimeout: time.Minute,
+	})
+
+	var dbtype protos.DBType
+	err := workflow.ExecuteActivity(checkCtx, snapshot.GetPeerType, name).Get(checkCtx, &dbtype)
+	return dbtype, err
+}
+
 func (s *SnapshotFlowExecution) setupReplication(
 	ctx workflow.Context,
 ) (*protos.SetupReplicationOutput, error) {

flow/workflows/util.go

@@ -1,8 +1,10 @@
 package peerflow
 
 import (
+	"log/slog"
 	"time"
 
+	"go.temporal.io/sdk/log"
 	"go.temporal.io/sdk/temporal"
 	"go.temporal.io/sdk/workflow"
 
@@ -26,14 +28,14 @@ func GetFlowMetadataContext(
 	ctx workflow.Context,
 	input *protos.FlowContextMetadataInput,
 ) (workflow.Context, error) {
-	metadataCtx := workflow.WithLocalActivityOptions(ctx, workflow.LocalActivityOptions{
+	metadataCtx := workflow.WithActivityOptions(ctx, workflow.ActivityOptions{
 		StartToCloseTimeout: time.Minute,
 		RetryPolicy:         &temporal.RetryPolicy{MaximumInterval: time.Minute},
 	})
-	getMetadataFuture := workflow.ExecuteLocalActivity(metadataCtx, flowable.GetFlowMetadata, input)
+	getMetadataFuture := workflow.ExecuteActivity(metadataCtx, flowable.GetFlowMetadata, input)
 	var metadata *protos.FlowContextMetadata
 	if err := getMetadataFuture.Get(metadataCtx, &metadata); err != nil {
-		return nil, err
+		return ctx, err
 	}
 	return workflow.WithValue(ctx, internal.FlowMetadataKey, metadata), nil
 }
@@ -43,3 +45,17 @@ func ShouldWorkflowContinueAsNew(ctx workflow.Context) bool {
 	return info.GetContinueAsNewSuggested() &&
 		(info.GetCurrentHistoryLength() > 40960 || info.GetCurrentHistorySize() > 40*1024*1024)
 }
+
+func getQRepOverwriteFullRefreshMode(wCtx workflow.Context, logger log.Logger, env map[string]string) bool {
+	checkCtx := workflow.WithActivityOptions(wCtx, workflow.ActivityOptions{
+		StartToCloseTimeout: time.Minute,
+	})
+
+	var fullRefreshEnabled bool
+	getFullRefreshFuture := workflow.ExecuteActivity(checkCtx, flowable.PeerDBFullRefreshOverwriteMode, env)
+	if err := getFullRefreshFuture.Get(checkCtx, &fullRefreshEnabled); err != nil {
+		logger.Warn("Failed to check if full refresh mode is enabled", slog.Any("error", err))
+		return false
+	}
+	return fullRefreshEnabled
+}