Clean up redis at the end of workflow

internal/benthos/benthos-builder/builders/sql-util.go

@@ -4,7 +4,6 @@ import (
 	"errors"
 	"fmt"
 	"log/slog"
-	"slices"
 	"strings"
 	"time"
 
@@ -13,7 +12,6 @@ import (
 	sqlmanager_shared "github.com/nucleuscloud/neosync/backend/pkg/sqlmanager/shared"
 	bb_internal "github.com/nucleuscloud/neosync/internal/benthos/benthos-builder/internal"
 	job_util "github.com/nucleuscloud/neosync/internal/job"
-	"github.com/nucleuscloud/neosync/internal/runconfigs"
 	neosync_benthos "github.com/nucleuscloud/neosync/worker/pkg/benthos"
 	"github.com/nucleuscloud/neosync/worker/pkg/workflows/datasync/activities/shared"
 )
@@ -421,25 +419,6 @@ func getColumnDefaultProperties(
 	return colDefaults, nil
 }
 
-func buildRedisDependsOnMap(transformedForeignKeyToSourceMap map[string][]*bb_internal.ReferenceKey, runconfig *runconfigs.RunConfig) map[string][]string {
-	redisDependsOnMap := map[string][]string{}
-	for col, fks := range transformedForeignKeyToSourceMap {
-		if !slices.Contains(runconfig.InsertColumns(), col) {
-			continue
-		}
-		for _, fk := range fks {
-			if _, exists := redisDependsOnMap[fk.Table]; !exists {
-				redisDependsOnMap[fk.Table] = []string{}
-			}
-			redisDependsOnMap[fk.Table] = append(redisDependsOnMap[fk.Table], fk.Column)
-		}
-	}
-	if runconfig.RunType() == runconfigs.RunTypeUpdate && len(redisDependsOnMap) != 0 {
-		redisDependsOnMap[runconfig.Table()] = runconfig.PrimaryKeys()
-	}
-	return redisDependsOnMap
-}
-
 type destinationOptions struct {
 	OnConflictDoNothing      bool
 	OnConflictDoUpdate       bool

internal/benthos/benthos-builder/builders/sql.go

@@ -256,11 +256,10 @@ func buildBenthosSqlSourceConfigResponses(
 		}
 
 		configs = append(configs, &bb_internal.BenthosSourceConfig{
-			Name:           config.Id(),
-			Config:         bc,
-			DependsOn:      config.DependsOn(),
-			RedisDependsOn: buildRedisDependsOnMap(transformedFktoPkMap, config),
-			RunType:        config.RunType(),
+			Name:      config.Id(),
+			Config:    bc,
+			DependsOn: config.DependsOn(),
+			RunType:   config.RunType(),
 
 			BenthosDsns: []*bb_shared.BenthosDsn{{ConnectionId: dsnConnectionId}},
 

worker/pkg/workflows/datasync/workflow/workflow.go

@@ -4,13 +4,11 @@ import (
 	"context"
 	"errors"
 	"fmt"
-	"slices"
 	"sync"
 	"time"
 
 	mgmtv1alpha1 "github.com/nucleuscloud/neosync/backend/gen/go/protos/mgmt/v1alpha1"
 	benthosbuilder "github.com/nucleuscloud/neosync/internal/benthos/benthos-builder"
-	benthosbuilder_shared "github.com/nucleuscloud/neosync/internal/benthos/benthos-builder/shared"
 	"github.com/nucleuscloud/neosync/internal/ee/license"
 	"github.com/nucleuscloud/neosync/internal/runconfigs"
 	neosync_benthos "github.com/nucleuscloud/neosync/worker/pkg/benthos"
@@ -177,15 +175,6 @@ func executeWorkflow(wfctx workflow.Context, req *WorkflowRequest) (*WorkflowRes
 		return nil, err
 	}
 
-	redisDependsOn := map[string]map[string][]string{} // schema.table -> dependson
-	redisConfigs := map[string]*benthosbuilder_shared.BenthosRedisConfig{}
-	for _, cfg := range bcResp.BenthosConfigs {
-		for _, redisCfg := range cfg.RedisConfig {
-			redisConfigs[redisCfg.Key] = redisCfg
-		}
-		redisDependsOn[cfg.Name] = cfg.RedisDependsOn
-	}
-
 	// spawn account status checker in loop
 	stopChan := workflow.NewNamedChannel(ctx, "account-status")
 	if initialCheckAccountStatusResponse.ShouldPoll {
@@ -276,9 +265,8 @@ func executeWorkflow(wfctx workflow.Context, req *WorkflowRequest) (*WorkflowRes
 				activityErr = err
 				cancelHandler()
 
-				// empty depends on map will clean up all redis inserts
 				detachedCtx, _ := workflow.NewDisconnectedContext(ctx)
-				redisErr := runRedisCleanUpActivity(detachedCtx, logger, map[string]map[string][]string{}, req.JobId, redisConfigs)
+				redisErr := runRedisCleanUpActivity(detachedCtx, logger, req.JobId, bcResp.BenthosConfigs)
 				if redisErr != nil {
 					logger.Error("redis clean up activity did not complete")
 				}
@@ -289,11 +277,6 @@ func executeWorkflow(wfctx workflow.Context, req *WorkflowRequest) (*WorkflowRes
 			if err != nil {
 				logger.Error(fmt.Sprintf("post table sync activity did not complete: %s", err.Error()), "schema", bc.TableSchema, "table", bc.TableName)
 			}
-			delete(redisDependsOn, bc.Name)
-			err = runRedisCleanUpActivity(ctx, logger, redisDependsOn, req.JobId, redisConfigs)
-			if err != nil {
-				logger.Error(fmt.Sprintf("redis clean up activity did not complete: %s", err))
-			}
 		})
 	}
 
@@ -390,6 +373,11 @@ func executeWorkflow(wfctx workflow.Context, req *WorkflowRequest) (*WorkflowRes
 		return nil, err
 	}
 
+	err = runRedisCleanUpActivity(ctx, logger, req.JobId, bcResp.BenthosConfigs)
+	if err != nil {
+		return nil, err
+	}
+
 	logger.Info("data sync workflow completed")
 	return &WorkflowResponse{}, nil
 }
@@ -520,16 +508,12 @@ func runPostTableSyncActivity(
 func runRedisCleanUpActivity(
 	ctx workflow.Context,
 	logger log.Logger,
-	dependsOnMap map[string]map[string][]string,
 	jobId string,
-	redisConfigs map[string]*benthosbuilder_shared.BenthosRedisConfig,
+	configs []*benthosbuilder.BenthosConfigResponse,
 ) error {
-	if len(redisConfigs) > 0 {
-		for k, cfg := range redisConfigs {
-			if !isReadyForCleanUp(cfg.Table, cfg.Column, dependsOnMap) {
-				continue
-			}
-			logger.Debug("executing redis clean up activity")
+	for _, cfg := range configs {
+		for _, redisCfg := range cfg.RedisConfig {
+			logger.Debug("executing redis clean up activity", "hashKey", redisCfg.Key)
 			var resp *syncrediscleanup_activity.DeleteRedisHashResponse
 			var redisCleanUpActivity *syncrediscleanup_activity.Activity
 			err := workflow.ExecuteActivity(
@@ -543,28 +527,16 @@ func runRedisCleanUpActivity(
 				redisCleanUpActivity.DeleteRedisHash,
 				&syncrediscleanup_activity.DeleteRedisHashRequest{
 					JobId:   jobId,
-					HashKey: cfg.Key,
+					HashKey: redisCfg.Key,
 				}).Get(ctx, &resp)
 			if err != nil {
 				return err
 			}
-			delete(redisConfigs, k)
 		}
 	}
 	return nil
 }
 
-func isReadyForCleanUp(table, col string, dependsOnMap map[string]map[string][]string) bool {
-	for _, dependsOn := range dependsOnMap {
-		for t, cols := range dependsOn {
-			if t == table && slices.Contains(cols, col) {
-				return false
-			}
-		}
-	}
-	return true
-}
-
 func withBenthosConfigResponseLoggerTags(bc *benthosbuilder.BenthosConfigResponse) []any {
 	keyvals := []any{}
 

worker/pkg/workflows/datasync/workflow/workflow_test.go

@@ -1006,34 +1006,6 @@ func Test_updateCompletedMap(t *testing.T) {
 	assert.Equal(t, []string{"name", "id"}, val)
 }
 
-func Test_isReadyForCleanUp(t *testing.T) {
-	assert.True(t, isReadyForCleanUp("", "", nil), "no dependencies")
-
-	assert.False(
-		t,
-		isReadyForCleanUp(
-			"table",
-			"col",
-			map[string]map[string][]string{
-				"other_table": {"table": []string{"col"}},
-			},
-		),
-		"has dependency",
-	)
-
-	assert.True(
-		t,
-		isReadyForCleanUp(
-			"table",
-			"col",
-			map[string]map[string][]string{
-				"other_table": {"table": []string{"col1"}},
-			},
-		),
-		"no dependency",
-	)
-}
-
 func Test_Workflow_Initial_AccountStatus(t *testing.T) {
 	testSuite := &testsuite.WorkflowTestSuite{}
 	env := testSuite.NewTestWorkflowEnvironment()