Auto refresh cache tweaks

flytepropeller/pkg/controller/nodes/subworkflow/handler.go

@@ -33,7 +33,7 @@
 }
 
 func (w *workflowNodeHandler) FinalizeRequired() bool {
-	return false
+	return true
 }
 
 func (w *workflowNodeHandler) Setup(_ context.Context, _ interfaces.SetupContext) error {
@@ -120,8 +120,12 @@
 	return nil
 }
 
-func (w *workflowNodeHandler) Finalize(ctx context.Context, _ interfaces.NodeExecutionContext) error {
-	logger.Warnf(ctx, "Subworkflow finalize invoked. Nothing to be done")
+func (w *workflowNodeHandler) Finalize(ctx context.Context, nCtx interfaces.NodeExecutionContext) error {
+	logger.Warnf(ctx, "Subworkflow finalize invoked. Clearing up the cache")
+	wfNode := nCtx.Node().GetWorkflowNode()
+	if wfNode.GetLaunchPlanRefID() != nil {
+		return w.lpHandler.HandleFinalize(ctx, nCtx)
+	}
 	return nil
 }
 

flytepropeller/pkg/controller/nodes/subworkflow/launchplan.go

@@ -229,3 +229,20 @@
 	}
 	return l.launchPlan.Kill(ctx, childID, fmt.Sprintf("cascading abort as parent execution id [%s] aborted, reason [%s]", nCtx.ExecutionContext().GetName(), reason))
 }
+
+func (l *launchPlanHandler) HandleFinalize(ctx context.Context, nCtx interfaces.NodeExecutionContext) error {
+	parentNodeExecutionID, err := getParentNodeExecutionID(nCtx)
+	if err != nil {
+		return err
+	}
+	childID, err := GetChildWorkflowExecutionIDForExecution(
+		parentNodeExecutionID,
+		nCtx,
+	)
+	if err != nil {
+		// THIS SHOULD NEVER HAPPEN
+		return err
+	}
+
+	return l.launchPlan.ClearCache(ctx, childID)
+}

flytepropeller/pkg/controller/nodes/subworkflow/launchplan/admin.go

@@ -224,6 +224,10 @@
 	return a.cache.Start(ctx)
 }
 
+func (a *adminLaunchPlanExecutor) ClearCache(ctx context.Context, executionID *core.WorkflowExecutionIdentifier) error {
+	return a.cache.DeleteDelayed(executionID.String())
+}
+
 func (a *adminLaunchPlanExecutor) syncItem(ctx context.Context, batch cache.Batch) (
 	resp []cache.ItemSyncResponse, err error) {
 	resp = make([]cache.ItemSyncResponse, 0, len(batch))

flytepropeller/pkg/controller/nodes/subworkflow/launchplan/launchplan.go

@@ -46,6 +46,9 @@ type Executor interface {
 
 	// Initialize initializes Executor.
 	Initialize(ctx context.Context) error
+
+	// ClearCache clears the cache of the LaunchPlan execution
+	ClearCache(ctx context.Context, executionID *core.WorkflowExecutionIdentifier) error
 }
 
 type Reader interface {