opensource/dify
2
0
Fork 0
mirror of https://github.com/langgenius/dify.git synced 2026-04-27 11:06:46 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat(workflow): enhance single run handling

Browse Source
This commit is contained in:
zhsama 2025-10-15 15:34:05 +08:00
parent 334e5f19bf
commit 112b5f63dd
4 changed files with 77 additions and 80 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

15
web/app/components/workflow/nodes/_base/components/workflow-panel/index.tsx
View File

@ -238,6 +238,7 @@ const BasePanel: FC<BasePanelProps> = ({
singleRunParams, singleRunParams,
nodeInfo, nodeInfo,
setRunInputData, setRunInputData,
handleStop,
handleSingleRun, handleSingleRun,
handleRunWithParams, handleRunWithParams,
getExistVarValuesInForms, getExistVarValuesInForms,
@ -438,18 +439,10 @@ const BasePanel: FC<BasePanelProps> = ({
<div <div
className='mr-1 flex h-6 w-6 cursor-pointer items-center justify-center rounded-md hover:bg-state-base-hover' className='mr-1 flex h-6 w-6 cursor-pointer items-center justify-center rounded-md hover:bg-state-base-hover'
onClick={() => { onClick={() => {
if (isSingleRunning) { if (isSingleRunning)
handleNodeDataUpdate({ handleStop()
id, else
data: {
_isSingleRun: false,
_singleRunningStatus: undefined,
},
})
}
else {
handleSingleRun() handleSingleRun()
}
}} }}
> >
{ {

111
web/app/components/workflow/nodes/_base/hooks/use-one-step-run.ts
View File

@ -205,6 +205,11 @@ const useOneStepRun = <T>({
const webhookSingleRunTimeoutRef = useRef<number | undefined>(undefined) const webhookSingleRunTimeoutRef = useRef<number | undefined>(undefined)
const webhookSingleRunTokenRef = useRef(0) const webhookSingleRunTokenRef = useRef(0)
const webhookSingleRunDelayResolveRef = useRef<(() => void) | null>(null) const webhookSingleRunDelayResolveRef = useRef<(() => void) | null>(null)
const pluginSingleRunActiveRef = useRef(false)
const pluginSingleRunAbortRef = useRef<AbortController | null>(null)
const pluginSingleRunTimeoutRef = useRef<number | undefined>(undefined)
const pluginSingleRunTokenRef = useRef(0)
const pluginSingleRunDelayResolveRef = useRef<(() => void) | null>(null)
const isPausedRef = useRef(isPaused) const isPausedRef = useRef(isPaused)
useEffect(() => { useEffect(() => {
isPausedRef.current = isPaused isPausedRef.current = isPaused
@ -263,6 +268,22 @@ const useOneStepRun = <T>({
} }
}, []) }, [])
const cancelPluginSingleRun = useCallback(() => {
pluginSingleRunActiveRef.current = false
pluginSingleRunTokenRef.current += 1
if (pluginSingleRunAbortRef.current)
pluginSingleRunAbortRef.current.abort()
pluginSingleRunAbortRef.current = null
if (pluginSingleRunTimeoutRef.current !== undefined) {
window.clearTimeout(pluginSingleRunTimeoutRef.current)
pluginSingleRunTimeoutRef.current = undefined
}
if (pluginSingleRunDelayResolveRef.current) {
pluginSingleRunDelayResolveRef.current()
pluginSingleRunDelayResolveRef.current = null
}
}, [])
const runWebhookSingleRun = useCallback(async (): Promise<any | null> => { const runWebhookSingleRun = useCallback(async (): Promise<any | null> => {
const urlPath = `/apps/${flowId}/workflows/draft/nodes/${id}/trigger/run` const urlPath = `/apps/${flowId}/workflows/draft/nodes/${id}/trigger/run`
const urlWithPrefix = `${API_PREFIX}${urlPath.startsWith('/') ? urlPath : `/${urlPath}`}` const urlWithPrefix = `${API_PREFIX}${urlPath.startsWith('/') ? urlPath : `/${urlPath}`}`
@ -335,7 +356,7 @@ const useOneStepRun = <T>({
data: { data: {
...data, ...data,
_isSingleRun: false, _isSingleRun: false,
_singleRunningStatus: NodeRunningStatus.Running, _singleRunningStatus: NodeRunningStatus.Listening,
}, },
}) })
@ -367,12 +388,12 @@ const useOneStepRun = <T>({
const urlPath = `/apps/${flowId}/workflows/draft/nodes/${id}/trigger/run` const urlPath = `/apps/${flowId}/workflows/draft/nodes/${id}/trigger/run`
const urlWithPrefix = `${API_PREFIX}${urlPath.startsWith('/') ? urlPath : `/${urlPath}`}` const urlWithPrefix = `${API_PREFIX}${urlPath.startsWith('/') ? urlPath : `/${urlPath}`}`
webhookSingleRunActiveRef.current = true pluginSingleRunActiveRef.current = true
const token = ++webhookSingleRunTokenRef.current const token = ++pluginSingleRunTokenRef.current
while (webhookSingleRunActiveRef.current && token === webhookSingleRunTokenRef.current) { while (pluginSingleRunActiveRef.current && token === pluginSingleRunTokenRef.current) {
const controller = new AbortController() const controller = new AbortController()
webhookSingleRunAbortRef.current = controller pluginSingleRunAbortRef.current = controller
try { try {
const baseOptions = getBaseOptions() const baseOptions = getBaseOptions()
@ -388,7 +409,7 @@ const useOneStepRun = <T>({
signal: controller.signal, signal: controller.signal,
}) })
if (!webhookSingleRunActiveRef.current || token !== webhookSingleRunTokenRef.current) if (!pluginSingleRunActiveRef.current || token !== pluginSingleRunTokenRef.current)
return null return null
const contentType = response.headers.get('Content-Type')?.toLowerCase() || '' const contentType = response.headers.get('Content-Type')?.toLowerCase() || ''
@ -397,35 +418,35 @@ const useOneStepRun = <T>({
if (!response.ok) { if (!response.ok) {
const message = responseData?.message || 'Plugin debug failed' const message = responseData?.message || 'Plugin debug failed'
Toast.notify({ type: 'error', message }) Toast.notify({ type: 'error', message })
cancelWebhookSingleRun() cancelPluginSingleRun()
throw new Error(message) throw new Error(message)
} }
if (responseData?.status === 'waiting') { if (responseData?.status === 'waiting') {
const delay = Number(responseData.retry_in) || 2000 const delay = Number(responseData.retry_in) || 2000
webhookSingleRunAbortRef.current = null pluginSingleRunAbortRef.current = null
if (!webhookSingleRunActiveRef.current || token !== webhookSingleRunTokenRef.current) if (!pluginSingleRunActiveRef.current || token !== pluginSingleRunTokenRef.current)
return null return null
await new Promise<void>((resolve) => { await new Promise<void>((resolve) => {
const timeoutId = window.setTimeout(resolve, delay) const timeoutId = window.setTimeout(resolve, delay)
webhookSingleRunTimeoutRef.current = timeoutId pluginSingleRunTimeoutRef.current = timeoutId
webhookSingleRunDelayResolveRef.current = resolve pluginSingleRunDelayResolveRef.current = resolve
controller.signal.addEventListener('abort', () => { controller.signal.addEventListener('abort', () => {
window.clearTimeout(timeoutId) window.clearTimeout(timeoutId)
resolve() resolve()
}, { once: true }) }, { once: true })
}) })
webhookSingleRunTimeoutRef.current = undefined pluginSingleRunTimeoutRef.current = undefined
webhookSingleRunDelayResolveRef.current = null pluginSingleRunDelayResolveRef.current = null
continue continue
} }
if (responseData?.status === 'error') { if (responseData?.status === 'error') {
const message = responseData.message || 'Plugin debug failed' const message = responseData.message || 'Plugin debug failed'
Toast.notify({ type: 'error', message }) Toast.notify({ type: 'error', message })
cancelWebhookSingleRun() cancelPluginSingleRun()
throw new Error(message) throw new Error(message)
} }
@ -434,33 +455,33 @@ const useOneStepRun = <T>({
data: { data: {
...data, ...data,
_isSingleRun: false, _isSingleRun: false,
_singleRunningStatus: NodeRunningStatus.Running, _singleRunningStatus: NodeRunningStatus.Listening,
}, },
}) })
cancelWebhookSingleRun() cancelPluginSingleRun()
return responseData return responseData
} }
catch (error) { catch (error) {
if (controller.signal.aborted && (!webhookSingleRunActiveRef.current || token !== webhookSingleRunTokenRef.current)) if (controller.signal.aborted && (!pluginSingleRunActiveRef.current || token !== pluginSingleRunTokenRef.current))
return null return null
if (controller.signal.aborted) if (controller.signal.aborted)
return null return null
console.error('handleRun: plugin debug polling error', error) console.error('handleRun: plugin debug polling error', error)
Toast.notify({ type: 'error', message: 'Plugin debug request failed' }) Toast.notify({ type: 'error', message: 'Plugin debug request failed' })
cancelWebhookSingleRun() cancelPluginSingleRun()
if (error instanceof Error) if (error instanceof Error)
throw error throw error
throw new Error(String(error)) throw new Error(String(error))
} }
finally { finally {
webhookSingleRunAbortRef.current = null pluginSingleRunAbortRef.current = null
} }
} }
return null return null
}, [flowId, id, data, handleNodeDataUpdate, cancelWebhookSingleRun]) }, [flowId, id, data, handleNodeDataUpdate, cancelPluginSingleRun])
const checkValidWrap = () => { const checkValidWrap = () => {
if (!checkValid) if (!checkValid)
@ -527,15 +548,17 @@ const useOneStepRun = <T>({
const isPluginNode = data.type === BlockEnum.TriggerPlugin const isPluginNode = data.type === BlockEnum.TriggerPlugin
const isTriggerNode = isWebhookNode || isPluginNode const isTriggerNode = isWebhookNode || isPluginNode
if (isTriggerNode) if (isWebhookNode)
cancelWebhookSingleRun() cancelWebhookSingleRun()
if (isPluginNode)
cancelPluginSingleRun()
handleNodeDataUpdate({ handleNodeDataUpdate({
id, id,
data: { data: {
...data, ...data,
_isSingleRun: false, _isSingleRun: false,
_singleRunningStatus: isTriggerNode ? NodeRunningStatus.Waiting : NodeRunningStatus.Running, _singleRunningStatus: isTriggerNode ? NodeRunningStatus.Listening : NodeRunningStatus.Running,
}, },
}) })
let res: any let res: any
@ -545,28 +568,32 @@ const useOneStepRun = <T>({
if (isWebhookNode) { if (isWebhookNode) {
res = await runWebhookSingleRun() res = await runWebhookSingleRun()
if (!res) { if (!res) {
handleNodeDataUpdate({ if (webhookSingleRunActiveRef.current) {
id, handleNodeDataUpdate({
data: { id,
...data, data: {
_isSingleRun: false, ...data,
_singleRunningStatus: NodeRunningStatus.NotStart, _isSingleRun: false,
}, _singleRunningStatus: NodeRunningStatus.Stopped,
}) },
})
}
return false return false
} }
} }
else if (isPluginNode) { else if (isPluginNode) {
res = await runPluginSingleRun() res = await runPluginSingleRun()
if (!res) { if (!res) {
handleNodeDataUpdate({ if (pluginSingleRunActiveRef.current) {
id, handleNodeDataUpdate({
data: { id,
...data, data: {
_isSingleRun: false, ...data,
_singleRunningStatus: NodeRunningStatus.NotStart, _isSingleRun: false,
}, _singleRunningStatus: NodeRunningStatus.Stopped,
}) },
})
}
return false return false
} }
} }
@ -817,8 +844,10 @@ const useOneStepRun = <T>({
} }
} }
finally { finally {
if (isTriggerNode) if (isWebhookNode)
cancelWebhookSingleRun() cancelWebhookSingleRun()
if (isPluginNode)
cancelPluginSingleRun()
if (!isPausedRef.current && !isIteration && !isLoop && res) { if (!isPausedRef.current && !isIteration && !isLoop && res) {
setRunResult({ setRunResult({
...res, ...res,
@ -846,11 +875,13 @@ const useOneStepRun = <T>({
const handleStop = () => { const handleStop = () => {
cancelWebhookSingleRun() cancelWebhookSingleRun()
cancelPluginSingleRun()
handleNodeDataUpdate({ handleNodeDataUpdate({
id, id,
data: { data: {
...data, ...data,
_singleRunningStatus: NodeRunningStatus.NotStart, _isSingleRun: false,
_singleRunningStatus: NodeRunningStatus.Stopped,
}, },
}) })
} }

30
web/app/components/workflow/run/status.tsx
View File

@ -26,34 +26,6 @@ const StatusPanel: FC<ResultProps> = ({
const docLink = useDocLink() const docLink = useDocLink()
const isListening = useStore(s => s.isListening) const isListening = useStore(s => s.isListening)
if (isListening) {
return (
<StatusContainer status={'running'}>
<div className='flex'>
<div className='max-w-[120px] flex-[33%]'>
<div className='system-2xs-medium-uppercase mb-1 text-text-tertiary'>{t('runLog.resultPanel.status')}</div>
<div className='system-xs-semibold-uppercase flex items-center gap-1 text-util-colors-blue-light-blue-light-600'>
<Indicator color='blue' />
<span>LISTENING</span>
</div>
</div>
<div className='max-w-[152px] flex-[33%]'>
<div className='system-2xs-medium-uppercase mb-1 text-text-tertiary'>{t('runLog.resultPanel.time')}</div>
<div className='system-sm-medium flex items-center gap-1 text-text-secondary'>
<div className='h-2 w-16 rounded-sm bg-text-quaternary' />
</div>
</div>
<div className='flex-[33%]'>
<div className='system-2xs-medium-uppercase mb-1 text-text-tertiary'>{t('runLog.resultPanel.tokens')}</div>
<div className='system-sm-medium flex items-center gap-1 text-text-secondary'>
<div className='h-2 w-20 rounded-sm bg-text-quaternary' />
</div>
</div>
</div>
</StatusContainer>
)
}
return ( return (
<StatusContainer status={status}> <StatusContainer status={status}>
<div className='flex'> <div className='flex'>
@ -75,7 +47,7 @@ const StatusPanel: FC<ResultProps> = ({
{status === 'running' && ( {status === 'running' && (
<> <>
<Indicator color={'blue'} /> <Indicator color={'blue'} />
<span>Running</span> <span>{isListening ? 'Listening' : 'Running'}</span>
</> </>
)} )}
{status === 'succeeded' && ( {status === 'succeeded' && (

1
web/app/components/workflow/types.ts
View File

@ -360,6 +360,7 @@ export enum WorkflowVersion {
export enum NodeRunningStatus { export enum NodeRunningStatus {
NotStart = 'not-start', NotStart = 'not-start',
Waiting = 'waiting', Waiting = 'waiting',
Listening = 'listening',
Running = 'running', Running = 'running',
Succeeded = 'succeeded', Succeeded = 'succeeded',
Failed = 'failed', Failed = 'failed',