PAUSE MIRROR support

flow/cmd/handler.go

@@ -154,7 +154,7 @@ func (h *FlowRequestHandler) CreateCDCFlow(
 		return nil, fmt.Errorf("unable to update flow config in catalog: %w", err)
 	}
 
-	state := peerflow.NewCDCFlowState()
+	state := peerflow.NewCDCFlowWorkflowState()
 	_, err = h.temporalClient.ExecuteWorkflow(
 		ctx,                                // context
 		workflowOptions,                    // workflow start options
@@ -350,6 +350,37 @@ func (h *FlowRequestHandler) ShutdownFlow(
 	}, nil
 }
 
+func (h *FlowRequestHandler) FlowStateChange(
+	ctx context.Context,
+	req *protos.FlowStateChangeRequest,
+) (*protos.FlowStateChangeResponse, error) {
+	var err error
+	if req.RequestedFlowState == protos.FlowState_STATE_PAUSED {
+		err = h.temporalClient.SignalWorkflow(
+			ctx,
+			req.WorkflowId,
+			"",
+			shared.CDCFlowSignalName,
+			shared.PauseSignal,
+		)
+	} else if req.RequestedFlowState == protos.FlowState_STATE_RUNNING {
+		err = h.temporalClient.SignalWorkflow(
+			ctx,
+			req.WorkflowId,
+			"",
+			shared.CDCFlowSignalName,
+			shared.NoopSignal,
+		)
+	}
+	if err != nil {
+		return nil, fmt.Errorf("unable to signal PeerFlow workflow: %w", err)
+	}
+
+	return &protos.FlowStateChangeResponse{
+		Ok: true,
+	}, nil
+}
+
 func (h *FlowRequestHandler) waitForWorkflowClose(ctx context.Context, workflowID string) error {
 	expBackoff := backoff.NewExponentialBackOff()
 	expBackoff.InitialInterval = 3 * time.Second

flow/connectors/postgres/postgres.go

@@ -2,7 +2,6 @@ package connpostgres
 
 import (
 	"context"
-	"database/sql"
 	"fmt"
 	"regexp"
 	"time"
@@ -839,7 +838,7 @@ func (c *PostgresConnector) SyncFlowCleanup(jobName string) error {
 	}
 	defer func() {
 		deferErr := syncFlowCleanupTx.Rollback(c.ctx)
-		if deferErr != sql.ErrTxDone && deferErr != nil {
+		if deferErr != pgx.ErrTxClosed && deferErr != nil {
 			log.WithFields(log.Fields{
 				"flowName": jobName,
 			}).Errorf("unexpected error while rolling back transaction for flow cleanup: %v", deferErr)

flow/e2e/test_utils.go

@@ -65,7 +65,7 @@ func SetupCDCFlowStatusQuery(env *testsuite.TestWorkflowEnvironment,
 			connectionGen.FlowJobName,
 		)
 		if err == nil {
-			var state peerflow.CDCFlowState
+			var state peerflow.CDCFlowWorkflowState
 			err = response.Get(&state)
 			if err != nil {
 				log.Errorln(err)
@@ -95,7 +95,7 @@ func NormalizeFlowCountQuery(env *testsuite.TestWorkflowEnvironment,
 			connectionGen.FlowJobName,
 		)
 		if err == nil {
-			var state peerflow.CDCFlowState
+			var state peerflow.CDCFlowWorkflowState
 			err = response.Get(&state)
 			if err != nil {
 				log.Errorln(err)

flow/shared/constants.go

@@ -12,6 +12,8 @@ type ContextKey string
 const (
 	NoopSignal CDCFlowSignal = iota
 	ShutdownSignal
+	PauseSignal
+
 	EnableMetricsKey    ContextKey = "enableMetrics"
 	CDCMirrorMonitorKey ContextKey = "cdcMirrorMonitor"
 )

flow/utils/signals.go

@@ -0,0 +1,27 @@
+package util
+
+import (
+	"github.com/PeerDB-io/peer-flow/shared"
+	"go.temporal.io/sdk/log"
+)
+
+func FlowSignalHandler(activeSignal shared.CDCFlowSignal,
+	v shared.CDCFlowSignal, logger log.Logger) shared.CDCFlowSignal {
+	if v == shared.ShutdownSignal {
+		logger.Info("received shutdown signal")
+		return v
+	} else if v == shared.PauseSignal {
+		logger.Info("received pause signal")
+		if activeSignal == shared.NoopSignal {
+			logger.Info("workflow was running, pausing it")
+			return v
+		}
+	} else if v == shared.NoopSignal {
+		logger.Info("received resume signal")
+		if activeSignal == shared.PauseSignal {
+			logger.Info("workflow was paused, resuming it")
+			return v
+		}
+	}
+	return activeSignal
+}