diff --git a/diagram-editor/dist.tar.gz b/diagram-editor/dist.tar.gz index 645cad48..a42acae3 100644 Binary files a/diagram-editor/dist.tar.gz and b/diagram-editor/dist.tar.gz differ diff --git a/diagram-editor/frontend/api.preprocessed.schema.json b/diagram-editor/frontend/api.preprocessed.schema.json index a57fbd11..76483c2f 100644 --- a/diagram-editor/frontend/api.preprocessed.schema.json +++ b/diagram-editor/frontend/api.preprocessed.schema.json @@ -849,49 +849,102 @@ } ] }, - "InteractionSessionMessage": { + "InteractionSessionFeedback": { "oneOf": [ { - "allOf": [ - { - "oneOf": [ - { - "properties": { - "operationStarted": { - "type": "string" - } - }, - "required": [ - "operationStarted" - ], - "type": "object" + "additionalProperties": false, + "properties": { + "operationStarted": { + "properties": { + "executionId": { + "type": "string" }, - { - "properties": { - "operationFinished": { - "type": "string" - } - }, - "required": [ - "operationFinished" - ], - "type": "object" + "operationId": { + "type": "string" } - ] - }, - { + }, + "required": [ + "operationId", + "executionId" + ], + "type": "object" + } + }, + "required": [ + "operationStarted" + ], + "type": "object" + }, + { + "additionalProperties": false, + "properties": { + "operationFinished": { "properties": { - "type": { - "const": "feedback", + "executionId": { + "type": "string" + }, + "operationId": { "type": "string" } }, "required": [ - "type" + "operationId", + "executionId" ], "type": "object" } - ] + }, + "required": [ + "operationFinished" + ], + "type": "object" + }, + { + "additionalProperties": false, + "properties": { + "connectionActivity": { + "properties": { + "sourceOperationId": { + "type": "string" + }, + "targetOperationId": { + "type": "string" + } + }, + "required": [ + "sourceOperationId", + "targetOperationId" + ], + "type": "object" + } + }, + "required": [ + "connectionActivity" + ], + "type": "object" + } + ] + }, + "InteractionSessionMessage": { + "oneOf": [ + { + "properties": { + "events": { + "items": { + "$ref": "#/$defs/InteractionSessionFeedback" + }, + "type": "array" + }, + "type": { + "const": "feedback", + "type": "string" + } + }, + "required": [ + "type", + "events" + ], + "type": "object" }, { "allOf": [ diff --git a/diagram-editor/frontend/app.css b/diagram-editor/frontend/app.css index 33eaa0cf..9bc48786 100644 --- a/diagram-editor/frontend/app.css +++ b/diagram-editor/frontend/app.css @@ -3,6 +3,16 @@ .react-flow.dark { --xy-edge-stroke-default: var(--mui-palette-grey-700); } +.react-flow__edge .react-flow__edge-path { + transition: + stroke 450ms ease-out, + filter 450ms ease-out; +} +.react-flow__edge.message-active .react-flow__edge-path { + stroke: var(--mui-palette-warning-dark); + filter: drop-shadow(0 0 4px var(--mui-palette-warning-dark)); + transition-duration: 0ms; +} .react-flow__pane.draggable { cursor: default; } @@ -34,24 +44,19 @@ background-color: var(--mui-palette-secondary-dark); } .react-flow__handle.handle-data-buffer { - background: linear-gradient( - -45deg, - var(--mui-palette-secondary-dark) 50%, - var(--xy-handle-background-color-default) 50% - ); -} -.react-flow__handle.handle-data-stream { - background: linear-gradient( - -45deg, - var(--mui-palette-warning-dark) 50%, - var(--xy-handle-background-color-default) 50% - ); + background-color: var(--mui-palette-secondary-main); } .react-flow__handle.connectionindicator { opacity: 1; border: 1px solid transparent; box-shadow: none; } +.react-flow__handle.handle-data-stream { + width: 10px; + height: 10px; + background-color: var(--mui-palette-background-default); + border: 2px solid var(--mui-palette-warning-dark); +} .react-flow__handle.handle-compatible { border: 1px solid var(--mui-palette-success-main); box-shadow: 0 0 0 4px rgb(76 175 80 / 22%); diff --git a/diagram-editor/frontend/diagram-editor.tsx b/diagram-editor/frontend/diagram-editor.tsx index 2d19c758..b93ab020 100644 --- a/diagram-editor/frontend/diagram-editor.tsx +++ b/diagram-editor/frontend/diagram-editor.tsx @@ -45,6 +45,7 @@ import { useDiagramProperties, } from './diagram-properties-provider'; import { useDiagramSidePanel } from './diagram-side-panel-controller'; +import { getEditPopoverPositionForNode } from './diagram-side-panel-layout'; import { clearDraftWorkspace, type DraftWorkspaceContent, @@ -66,6 +67,7 @@ import { EditEdgeForm, EditNodeForm } from './forms'; import EditScopeForm from './forms/edit-scope-form'; import type { ScriptNodeEnvironmentBinding } from './forms/script-environment-workspace'; import { useScriptEnvironmentNavigation } from './forms/use-script-environment-navigation'; +import { glowEdge } from './handles'; import { type InteractionVisualizationContext, InteractionVisualizationProvider, @@ -82,6 +84,7 @@ import { MaterialSymbol, NODE_TYPES, type OperationNode, + START_ID, TERMINATE_ID, } from './nodes'; import { NotificationProvider } from './notification-provider'; @@ -248,6 +251,22 @@ function getInteractionNodeId( ? operationId.slice(1) : operationId; + if (normalizedId === '(start)') { + return ( + nodeManager.tryGetNode(joinNamespaces(ROOT_NAMESPACE, START_ID))?.id ?? + null + ); + } + + if (normalizedId.endsWith(':(start)')) { + const namespace = normalizedId.slice(0, -':(start)'.length); + return ( + nodeManager.tryGetNode( + joinNamespaces(ROOT_NAMESPACE, namespace, START_ID), + )?.id ?? null + ); + } + if (normalizedId === '(terminate)') { return ( nodeManager.tryGetNode(joinNamespaces(ROOT_NAMESPACE, TERMINATE_ID)) @@ -303,21 +322,32 @@ function DiagramEditor() { React.useState(() => new Set()); const [interactionVisitedNodeIds, setInteractionVisitedNodeIds] = React.useState(() => new Set()); + const interactionExecutionNodeIds = React.useRef(new Map()); const clearInteractionVisualization = React.useCallback(() => { + interactionExecutionNodeIds.current.clear(); setInteractionActiveNodeIds(new Set()); setInteractionVisitedNodeIds(new Set()); }, []); const markInteractionFinished = React.useCallback(() => { + interactionExecutionNodeIds.current.clear(); setInteractionActiveNodeIds(new Set()); }, []); const markInteractionOperationFinished = React.useCallback( - (operationId: string) => { - const nodeId = getInteractionNodeId(operationId, nodeManager); + (operationId: string, executionId: string) => { + const nodeId = + interactionExecutionNodeIds.current.get(executionId) ?? + getInteractionNodeId(operationId, nodeManager); if (!nodeId) { return; } + interactionExecutionNodeIds.current.delete(executionId); setInteractionActiveNodeIds((prev) => { + if ( + [...interactionExecutionNodeIds.current.values()].includes(nodeId) + ) { + return prev; + } const next = new Set(prev); next.delete(nodeId); return next; @@ -331,12 +361,13 @@ function DiagramEditor() { [nodeManager], ); const markInteractionOperationStarted = React.useCallback( - (operationId: string) => { + (operationId: string, executionId: string) => { const nodeId = getInteractionNodeId(operationId, nodeManager); if (!nodeId) { return; } + interactionExecutionNodeIds.current.set(executionId, nodeId); setInteractionVisitedNodeIds((prev) => { const next = new Set(prev); next.delete(nodeId); @@ -350,6 +381,23 @@ function DiagramEditor() { }, [nodeManager], ); + const markInteractionConnection = React.useCallback( + (sourceOperationId: string, targetOperationId: string) => { + const sourceNodeId = getInteractionNodeId(sourceOperationId, nodeManager); + const targetNodeId = getInteractionNodeId(targetOperationId, nodeManager); + if (!sourceNodeId || !targetNodeId) { + return; + } + reactFlowInstance.current + ?.getEdges() + .filter( + (edge) => + edge.source === sourceNodeId && edge.target === targetNodeId, + ) + .forEach((edge) => glowEdge(edge.id)); + }, + [nodeManager], + ); const interactionVisualizationContext = React.useMemo( () => ({ @@ -359,6 +407,7 @@ function DiagramEditor() { markInteractionFinished, markInteractionOperationFinished, markInteractionOperationStarted, + markInteractionConnection, }), [ clearInteractionVisualization, @@ -367,6 +416,7 @@ function DiagramEditor() { markInteractionFinished, markInteractionOperationFinished, markInteractionOperationStarted, + markInteractionConnection, ], ); const savedNodes = React.useRef([]); @@ -387,7 +437,11 @@ function DiagramEditor() { } = useTransientEditorDrafts(); const openScriptEnvironment = useScriptEnvironmentNavigation(); const { - state: { open: sidePanelOpen, tab: sidePanelTab }, + state: { + open: sidePanelOpen, + expanded: sidePanelExpanded, + tab: sidePanelTab, + }, } = useDiagramSidePanel(); const updateEditorModeAction = React.useCallback( @@ -761,7 +815,6 @@ function DiagramEditor() { handleNodeChange(change); closeAllPopovers(); }; - if (node.type === 'scope') { return ( { diff --git a/diagram-editor/frontend/diagram-side-panel-layout.test.ts b/diagram-editor/frontend/diagram-side-panel-layout.test.ts index 9244c3a8..8cb6945f 100644 --- a/diagram-editor/frontend/diagram-side-panel-layout.test.ts +++ b/diagram-editor/frontend/diagram-side-panel-layout.test.ts @@ -1,6 +1,7 @@ import { constrainEditPopoverPosition, getDiagramSidePanelWidth, + getEditPopoverPositionForNode, } from './diagram-side-panel-layout'; describe('diagram side-panel layout', () => { @@ -68,4 +69,24 @@ describe('diagram side-panel layout', () => { }), ).toEqual({ left: 16, top: 300 }); }); + + test('places the editor right of a node when it fits', () => { + expect( + getEditPopoverPositionForNode({ + nodeRect: { left: 200, right: 242, top: 300 }, + viewportWidth: 1440, + sidePanel: { open: false, expanded: false }, + }), + ).toEqual({ left: 258, top: 300 }); + }); + + test('flips the editor left of a node when the drawer leaves no room', () => { + expect( + getEditPopoverPositionForNode({ + nodeRect: { left: 600, right: 642, top: 300 }, + viewportWidth: 1440, + sidePanel: { open: true, expanded: true }, + }), + ).toEqual({ left: 164, top: 300 }); + }); }); diff --git a/diagram-editor/frontend/diagram-side-panel-layout.ts b/diagram-editor/frontend/diagram-side-panel-layout.ts index 55a92641..8bf60a8a 100644 --- a/diagram-editor/frontend/diagram-side-panel-layout.ts +++ b/diagram-editor/frontend/diagram-side-panel-layout.ts @@ -1,13 +1,10 @@ +import type { PopoverPosition } from '@mui/material'; + export interface SidePanelLayoutState { open: boolean; expanded: boolean; } -export interface EditorAnchorPosition { - left: number; - top: number; -} - export const NormalSidePanelWidth = 420; export const ExpandedSidePanelWidth = 900; export const MinimumVisibleCanvasWidth = 56; @@ -29,21 +26,30 @@ export function getDiagramSidePanelWidth( ); } +/** Right edge of the area the edit popover may occupy. */ +function getEditPopoverRightBoundary( + viewportWidth: number, + sidePanel: SidePanelLayoutState, +): number { + const sidePanelWidth = getDiagramSidePanelWidth(viewportWidth, sidePanel); + return ( + viewportWidth - + sidePanelWidth - + (sidePanelWidth > 0 ? EditPopoverSidePanelGap : EditPopoverMargin) + ); +} + export function constrainEditPopoverPosition({ anchorPosition, viewportWidth, sidePanel, }: { - anchorPosition: EditorAnchorPosition; + anchorPosition: PopoverPosition; viewportWidth: number; sidePanel: SidePanelLayoutState; -}): EditorAnchorPosition { - const sidePanelWidth = getDiagramSidePanelWidth(viewportWidth, sidePanel); - const rightBoundary = - viewportWidth - - sidePanelWidth - - (sidePanelWidth > 0 ? EditPopoverSidePanelGap : EditPopoverMargin); - const maximumLeft = rightBoundary - EditPopoverWidth; +}): PopoverPosition { + const maximumLeft = + getEditPopoverRightBoundary(viewportWidth, sidePanel) - EditPopoverWidth; return { left: Math.max( @@ -53,3 +59,26 @@ export function constrainEditPopoverPosition({ top: anchorPosition.top, }; } + +/** Places the edit popover beside a node, flipping to its left when it would not fit. */ +export function getEditPopoverPositionForNode({ + nodeRect, + viewportWidth, + sidePanel, +}: { + nodeRect: Pick; + viewportWidth: number; + sidePanel: SidePanelLayoutState; +}): PopoverPosition { + const rightOfNode = nodeRect.right + EditPopoverMargin; + const fitsRight = + rightOfNode + EditPopoverWidth <= + getEditPopoverRightBoundary(viewportWidth, sidePanel); + + return { + left: fitsRight + ? rightOfNode + : nodeRect.left - EditPopoverWidth - EditPopoverMargin, + top: nodeRect.top, + }; +} diff --git a/diagram-editor/frontend/handles.test.tsx b/diagram-editor/frontend/handles.test.tsx new file mode 100644 index 00000000..66470449 --- /dev/null +++ b/diagram-editor/frontend/handles.test.tsx @@ -0,0 +1,29 @@ +import { glowEdge } from './handles'; + +describe('edge activity', () => { + beforeEach(() => { + jest.useFakeTimers(); + document.body.innerHTML = + ''; + }); + + afterEach(() => { + jest.useRealTimers(); + document.body.replaceChildren(); + }); + + test('keeps an edge glowing while messages remain frequent', () => { + const edge = document.querySelector('[data-id="edge-1"]'); + + glowEdge('edge-1'); + expect(edge).toHaveClass('message-active'); + + jest.advanceTimersByTime(150); + glowEdge('edge-1'); + jest.advanceTimersByTime(150); + expect(edge).toHaveClass('message-active'); + + jest.advanceTimersByTime(50); + expect(edge).not.toHaveClass('message-active'); + }); +}); diff --git a/diagram-editor/frontend/handles.tsx b/diagram-editor/frontend/handles.tsx index a9a5c8ee..8538539a 100644 --- a/diagram-editor/frontend/handles.tsx +++ b/diagram-editor/frontend/handles.tsx @@ -29,6 +29,29 @@ export interface HandleProps extends Omit { variant: HandleType; } +const EdgeMessageIdleMs = 200; +const edgeMessageTimers = new WeakMap(); + +export function glowEdge(edgeId: string) { + const edge = document.querySelector(`.react-flow__edge[data-id="${edgeId}"]`); + if (!edge) { + return; + } + + edge.classList.add('message-active'); + const timer = edgeMessageTimers.get(edge); + if (timer !== undefined) { + window.clearTimeout(timer); + } + edgeMessageTimers.set( + edge, + window.setTimeout(() => { + edge.classList.remove('message-active'); + edgeMessageTimers.delete(edge); + }, EdgeMessageIdleMs), + ); +} + function variantClassName(handleType?: HandleType): string | undefined { if (handleType === undefined) { return undefined; diff --git a/diagram-editor/frontend/interaction-visualization-provider.tsx b/diagram-editor/frontend/interaction-visualization-provider.tsx index 252df427..052448d4 100644 --- a/diagram-editor/frontend/interaction-visualization-provider.tsx +++ b/diagram-editor/frontend/interaction-visualization-provider.tsx @@ -5,8 +5,18 @@ export interface InteractionVisualizationContext { visitedNodeIds: Set; clearInteractionVisualization: () => void; markInteractionFinished: () => void; - markInteractionOperationFinished: (operationId: string) => void; - markInteractionOperationStarted: (operationId: string) => void; + markInteractionOperationStarted: ( + operationId: string, + executionId: string, + ) => void; + markInteractionOperationFinished: ( + operationId: string, + executionId: string, + ) => void; + markInteractionConnection: ( + sourceOperationId: string, + targetOperationId: string, + ) => void; } const DefaultInteractionVisualizationContext: InteractionVisualizationContext = @@ -15,8 +25,9 @@ const DefaultInteractionVisualizationContext: InteractionVisualizationContext = visitedNodeIds: new Set(), clearInteractionVisualization: () => {}, markInteractionFinished: () => {}, - markInteractionOperationFinished: () => {}, markInteractionOperationStarted: () => {}, + markInteractionOperationFinished: () => {}, + markInteractionConnection: () => {}, }; const InteractionVisualizationContextComp = diff --git a/diagram-editor/frontend/nodes/base-node.tsx b/diagram-editor/frontend/nodes/base-node.tsx index c59bb8bb..a1b23b04 100644 --- a/diagram-editor/frontend/nodes/base-node.tsx +++ b/diagram-editor/frontend/nodes/base-node.tsx @@ -12,6 +12,8 @@ import { type JSX, memo } from 'react'; import { useInteractionVisualization } from '../interaction-visualization-provider'; import { LAYOUT_OPTIONS } from '../utils/layout'; +const CompactNodeSize = 42; + export interface BaseNodeProps extends NodeProps { color?: ButtonProps['color']; icon?: React.JSX.Element | string; @@ -19,6 +21,7 @@ export interface BaseNodeProps extends NodeProps { caption?: string; handles?: JSX.Element; highlight?: boolean; + compact?: boolean; } function BaseNode({ @@ -30,6 +33,7 @@ function BaseNode({ selected, id, highlight, + compact, }: BaseNodeProps) { const { activeNodeIds, visitedNodeIds } = useInteractionVisualization(); const interactionActive = activeNodeIds.has(id); @@ -44,6 +48,7 @@ function BaseNode({ return ( ({ + borderRadius: compact ? '50%' : undefined, outline: interactionActive ? `2px solid ${theme.palette.success.main}` : interactionVisited @@ -62,6 +67,18 @@ function BaseNode({ `0 0 8px 3px ${alpha(theme.palette.info.main, 0.35)}`, ].join(', ') : undefined, + animation: interactionVisited + ? 'interaction-finished 450ms ease-out' + : undefined, + '@keyframes interaction-finished': { + from: { + outlineColor: theme.palette.success.main, + boxShadow: [ + `0 0 0 4px ${alpha(theme.palette.success.main, 0.28)}`, + `0 0 18px 6px ${alpha(theme.palette.success.main, 0.35)}`, + ].join(', '), + }, + }, transition: theme.transitions.create(['box-shadow', 'outline-color'], { duration: theme.transitions.duration.shortest, }), @@ -69,32 +86,33 @@ function BaseNode({ > {handles} diff --git a/diagram-editor/frontend/nodes/fork-clone-node.tsx b/diagram-editor/frontend/nodes/fork-clone-node.tsx index 34994c1b..6d0d2cb5 100644 --- a/diagram-editor/frontend/nodes/fork-clone-node.tsx +++ b/diagram-editor/frontend/nodes/fork-clone-node.tsx @@ -8,6 +8,7 @@ function ForkCloneNodeComp(props: NodeProps>) { return ( } label="Fork Clone" handles={ diff --git a/diagram-editor/frontend/nodes/fork-result-node.tsx b/diagram-editor/frontend/nodes/fork-result-node.tsx index 0113e3c3..a914ab2f 100644 --- a/diagram-editor/frontend/nodes/fork-result-node.tsx +++ b/diagram-editor/frontend/nodes/fork-result-node.tsx @@ -7,17 +7,18 @@ import { ForkResultIcon } from './icons'; const ForkResultOkHandle = styled(Handle)(({ theme }) => ({ left: '25%', - background: `linear-gradient(-45deg, ${theme.palette.success.main} 50%, var(--xy-handle-background-color-default) 50%)`, + backgroundColor: theme.palette.success.main, })); const ForkResultErrHandle = styled(Handle)(({ theme }) => ({ left: '75%', - background: `linear-gradient(-45deg, ${theme.palette.error.main} 50%, var(--xy-handle-background-color-default) 50%)`, + backgroundColor: theme.palette.error.main, })); function ForkResultNodeComp(props: NodeProps>) { return ( } label="Fork Result" handles={ diff --git a/diagram-editor/frontend/nodes/icons.tsx b/diagram-editor/frontend/nodes/icons.tsx index 31c96130..3668cd79 100644 --- a/diagram-editor/frontend/nodes/icons.tsx +++ b/diagram-editor/frontend/nodes/icons.tsx @@ -31,6 +31,10 @@ export function ForkCloneIcon(): React.JSX.Element { return ; } +export function ForkResultIcon(): React.JSX.Element { + return ; +} + export function TransformIcon(): React.JSX.Element { return ; } @@ -53,16 +57,14 @@ export function SplitIcon(): React.JSX.Element { ); } -export function ForkResultIcon(): React.JSX.Element { - return ; -} - export function ListenIcon(): React.JSX.Element { return ; } export function JoinIcon(): React.JSX.Element { - return ; + return ( + + ); } export function StreamOutIcon(): React.JSX.Element { diff --git a/diagram-editor/frontend/nodes/join-node.tsx b/diagram-editor/frontend/nodes/join-node.tsx index b650688f..07eb3f10 100644 --- a/diagram-editor/frontend/nodes/join-node.tsx +++ b/diagram-editor/frontend/nodes/join-node.tsx @@ -8,6 +8,7 @@ function JoinNodeComp(props: NodeProps>) { return ( } label="Join" handles={ diff --git a/diagram-editor/frontend/nodes/split-node.tsx b/diagram-editor/frontend/nodes/split-node.tsx index 8949f303..44e02c7f 100644 --- a/diagram-editor/frontend/nodes/split-node.tsx +++ b/diagram-editor/frontend/nodes/split-node.tsx @@ -8,6 +8,7 @@ function SplitNodeComp(props: NodeProps>) { return ( } label="Split" handles={ diff --git a/diagram-editor/frontend/nodes/unzip-node.tsx b/diagram-editor/frontend/nodes/unzip-node.tsx index d976da61..78b28fa8 100644 --- a/diagram-editor/frontend/nodes/unzip-node.tsx +++ b/diagram-editor/frontend/nodes/unzip-node.tsx @@ -8,6 +8,7 @@ function UnzipNodeComp(props: NodeProps>) { return ( } label="Unzip" handles={ diff --git a/diagram-editor/frontend/run-button.tsx b/diagram-editor/frontend/run-button.tsx index 62857bab..432fe5df 100644 --- a/diagram-editor/frontend/run-button.tsx +++ b/diagram-editor/frontend/run-button.tsx @@ -18,7 +18,11 @@ import { useNodeManager } from './node-manager'; import { MaterialSymbol } from './nodes'; import { useRegistry } from './registry-provider'; import { useTemplates } from './templates-provider'; -import type { Diagram, DiagramOperation } from './types/api'; +import type { + Diagram, + DiagramOperation, + InteractionSessionFeedback, +} from './types/api'; import { useEdges } from './use-edges'; import { exportDiagram } from './utils/export-diagram'; @@ -32,6 +36,7 @@ interface ExecutionTimelineEntry { const DefaultResponseContent: ResponseContent = { raw: '' }; const MaxExecutionTimelineEntries = 200; +const MaxInteractionPlaybackFrames = 60; function enableInteractionTraceForOps(ops: Record) { for (const op of Object.values(ops)) { @@ -68,6 +73,7 @@ export function RunPanel({ const { clearInteractionVisualization, markInteractionFinished, + markInteractionConnection, markInteractionOperationFinished, markInteractionOperationStarted, } = useInteractionVisualization(); @@ -84,6 +90,9 @@ export function RunPanel({ ReturnType> > | null>(null); const interactionSubscriptionRef = useRef(null); + const interactionPlaybackQueue = useRef([]); + const interactionPlaybackFrame = useRef(null); + const interactionFinishPending = useRef(false); const [diagramProperties] = useDiagramProperties(); const closeInteractionSession = useCallback(() => { @@ -93,17 +102,70 @@ export function RunPanel({ interactionSessionRef.current = null; }, []); + const clearInteractionPlayback = useCallback(() => { + if (interactionPlaybackFrame.current !== null) { + cancelAnimationFrame(interactionPlaybackFrame.current); + } + interactionPlaybackFrame.current = null; + interactionPlaybackQueue.current.length = 0; + interactionFinishPending.current = false; + }, []); + + const playInteractionEvents = useCallback( + function play() { + const queue = interactionPlaybackQueue.current; + const events = queue.splice( + 0, + Math.max(1, Math.ceil(queue.length / MaxInteractionPlaybackFrames)), + ); + + for (const event of events) { + if ('operationStarted' in event) { + const { operationId, executionId } = event.operationStarted; + markInteractionOperationStarted(operationId, executionId); + } else if ('operationFinished' in event) { + const { operationId, executionId } = event.operationFinished; + markInteractionOperationFinished(operationId, executionId); + } else if ('connectionActivity' in event) { + const { sourceOperationId, targetOperationId } = + event.connectionActivity; + markInteractionConnection(sourceOperationId, targetOperationId); + } + } + + if (queue.length > 0) { + interactionPlaybackFrame.current = requestAnimationFrame(play); + } else { + interactionPlaybackFrame.current = null; + if (interactionFinishPending.current) { + interactionFinishPending.current = false; + markInteractionFinished(); + } + } + }, + [ + markInteractionConnection, + markInteractionFinished, + markInteractionOperationFinished, + markInteractionOperationStarted, + ], + ); + useEffect(() => { - return closeInteractionSession; - }, [closeInteractionSession]); + return () => { + clearInteractionPlayback(); + closeInteractionSession(); + }; + }, [clearInteractionPlayback, closeInteractionSession]); useEffect(() => { showProgressRef.current = showProgress; if (!showProgress) { + clearInteractionPlayback(); clearInteractionVisualization(); setExecutionTimeline([]); } - }, [clearInteractionVisualization, showProgress]); + }, [clearInteractionPlayback, clearInteractionVisualization, showProgress]); const requestError = useMemo(() => { try { @@ -143,6 +205,7 @@ export function RunPanel({ }; const handleRunClick = async () => { + clearInteractionPlayback(); closeInteractionSession(); clearInteractionVisualization(); setExecutionTimeline([]); @@ -173,40 +236,40 @@ export function RunPanel({ interactionSubscriptionRef.current = interactionSession.interactionMessages$.subscribe({ next: (msg) => { - if ( - msg.type === 'feedback' && - 'operationStarted' in msg && - typeof msg.operationStarted === 'string' - ) { + if (msg.type === 'feedback') { if (!showProgressRef.current) { return; } - const operationId = msg.operationStarted; - markInteractionOperationStarted(operationId); - const entry = { - seq: ++interactionEventCounter.current, - operationId, - }; - setExecutionTimeline((prev) => - [...prev, entry].slice(-MaxExecutionTimelineEntries), - ); - return; - } - - if ( - msg.type === 'feedback' && - 'operationFinished' in msg && - typeof msg.operationFinished === 'string' - ) { - if (!showProgressRef.current) { - return; + const entries: ExecutionTimelineEntry[] = []; + for (const event of msg.events) { + if ('operationStarted' in event) { + const { operationId } = event.operationStarted; + entries.push({ + seq: ++interactionEventCounter.current, + operationId, + }); + } + } + if (entries.length > 0) { + setExecutionTimeline((prev) => + [...prev, ...entries].slice(-MaxExecutionTimelineEntries), + ); + } + interactionPlaybackQueue.current.push(...msg.events); + if (interactionPlaybackFrame.current === null) { + interactionPlaybackFrame.current = requestAnimationFrame( + playInteractionEvents, + ); } - markInteractionOperationFinished(msg.operationFinished); return; } if (msg.type === 'finish') { - markInteractionFinished(); + if (interactionPlaybackFrame.current !== null) { + interactionFinishPending.current = true; + } else { + markInteractionFinished(); + } if ('ok' in msg) { setResponseContent({ raw: JSON.stringify(msg.ok, null, 2) }); } else { @@ -217,6 +280,7 @@ export function RunPanel({ } }, error: (err) => { + clearInteractionPlayback(); markInteractionFinished(); setResponseContent({ err: (err as Error).message }); setRunningMode(null); diff --git a/diagram-editor/frontend/types/api.d.ts b/diagram-editor/frontend/types/api.d.ts index ab32ab1a..b65db699 100644 --- a/diagram-editor/frontend/types/api.d.ts +++ b/diagram-editor/frontend/types/api.d.ts @@ -252,24 +252,42 @@ export type Schema = [k: string]: unknown; } | boolean; +/** + * This interface was referenced by `DiagramEditorApi`'s JSON-Schema + * via the `definition` "InteractionSessionFeedback". + */ +export type InteractionSessionFeedback = + | { + operationStarted: { + executionId: string; + operationId: string; + [k: string]: unknown; + }; + } + | { + operationFinished: { + executionId: string; + operationId: string; + [k: string]: unknown; + }; + } + | { + connectionActivity: { + sourceOperationId: string; + targetOperationId: string; + [k: string]: unknown; + }; + }; /** * This interface was referenced by `DiagramEditorApi`'s JSON-Schema * via the `definition` "InteractionSessionMessage". */ export type InteractionSessionMessage = - | (( - | { - operationStarted: string; - [k: string]: unknown; - } - | { - operationFinished: string; - [k: string]: unknown; - } - ) & { + | { + events: InteractionSessionFeedback[]; type: 'feedback'; [k: string]: unknown; - }) + } | (( | { ok: unknown; diff --git a/diagram-editor/frontend/use-responsive-edit-popover-position.ts b/diagram-editor/frontend/use-responsive-edit-popover-position.ts index 119adcae..2571d687 100644 --- a/diagram-editor/frontend/use-responsive-edit-popover-position.ts +++ b/diagram-editor/frontend/use-responsive-edit-popover-position.ts @@ -1,6 +1,6 @@ +import type { PopoverPosition } from '@mui/material'; import { useEffect, useMemo, useState } from 'react'; import { useDiagramSidePanel } from './diagram-side-panel-controller'; -import type { EditorAnchorPosition } from './diagram-side-panel-layout'; import { constrainEditPopoverPosition, getDiagramSidePanelWidth, @@ -33,8 +33,8 @@ export function useDiagramSidePanelWidth(): number { } export function useResponsiveEditPopoverPosition( - anchorPosition?: EditorAnchorPosition, -): EditorAnchorPosition | undefined { + anchorPosition?: PopoverPosition, +): PopoverPosition | undefined { const { state: { open, expanded }, } = useDiagramSidePanel(); diff --git a/diagram-editor/server/api/executor.rs b/diagram-editor/server/api/executor.rs index 8613084c..b3861abc 100644 --- a/diagram-editor/server/api/executor.rs +++ b/diagram-editor/server/api/executor.rs @@ -26,16 +26,11 @@ use std::{ }; use tokio::sync::mpsc::error::TryRecvError; use tracing::error; -#[cfg(feature = "router")] -use tracing::warn; #[cfg(feature = "router")] use super::websocket::{WebsocketSinkExt, WebsocketStreamExt}; use crate::api::error_responses::WorkflowCancelledResponse; -#[cfg(feature = "router")] -type BroadcastRecvError = tokio::sync::broadcast::error::RecvError; - type WorkflowResponseResult = Result<(Outcome, Entity), Box>; type WorkflowResponseSender = tokio::sync::oneshot::Sender; @@ -43,7 +38,7 @@ type WorkflowResponseSender = tokio::sync::oneshot::Sender); +struct FeedbackSender(tokio::sync::mpsc::UnboundedSender); pub struct Context { diagram: Diagram, @@ -889,10 +884,20 @@ impl InteractionSessionEnd { #[cfg_attr(feature = "json_schema", derive(schemars::JsonSchema))] #[cfg_attr(test, derive(serde::Deserialize))] #[derive(Serialize)] -#[serde(rename_all = "camelCase")] +#[serde(rename_all = "camelCase", rename_all_fields = "camelCase")] pub enum InteractionSessionFeedback { - OperationStarted(String), - OperationFinished(String), + OperationStarted { + operation_id: String, + execution_id: String, + }, + OperationFinished { + operation_id: String, + execution_id: String, + }, + ConnectionActivity { + source_operation_id: String, + target_operation_id: String, + }, } #[cfg_attr(feature = "json_schema", derive(schemars::JsonSchema))] @@ -900,7 +905,9 @@ pub enum InteractionSessionFeedback { #[derive(Serialize)] #[serde(rename_all = "camelCase", tag = "type")] pub enum InteractionSessionMessage { - Feedback(InteractionSessionFeedback), + Feedback { + events: Vec, + }, Finish(InteractionSessionEnd), } @@ -919,7 +926,7 @@ where }; let (response_tx, response_rx) = tokio::sync::oneshot::channel(); - let (feedback_tx, mut feedback_rx) = tokio::sync::broadcast::channel(10); + let (feedback_tx, mut feedback_rx) = tokio::sync::mpsc::unbounded_channel(); if let Err(err) = state .send_chan .send(Context { @@ -981,18 +988,10 @@ where tokio::select! { feedback = feedback_rx.recv() => { match feedback { - Ok(feedback) => { - send_interaction_feedback(&mut write, &feedback).await; + Some(feedback) => { + send_interaction_feedback(&mut write, feedback, &mut feedback_rx).await; } - Err(e) => match e { - BroadcastRecvError::Closed => { - feedback_open = false; - } - BroadcastRecvError::Lagged(_) => { - warn!("{}", e); - feedback_open = false; - } - }, + None => feedback_open = false, } } result = &mut response => { @@ -1016,77 +1015,119 @@ where #[cfg(feature = "router")] async fn drain_interaction_feedback( write: &mut W, - feedback_rx: &mut tokio::sync::broadcast::Receiver, + feedback_rx: &mut tokio::sync::mpsc::UnboundedReceiver, ) where W: WebsocketSinkExt, { - loop { - match feedback_rx.try_recv() { - Ok(feedback) => send_interaction_feedback(write, &feedback).await, - Err(tokio::sync::broadcast::error::TryRecvError::Empty) => break, - Err(tokio::sync::broadcast::error::TryRecvError::Closed) => break, - Err(tokio::sync::broadcast::error::TryRecvError::Lagged(skipped)) => { - warn!("interaction feedback lagged by {skipped} messages"); - } - } + while let Ok(feedback) = feedback_rx.try_recv() { + send_interaction_feedback(write, feedback, feedback_rx).await; } } #[cfg(feature = "router")] -async fn send_interaction_feedback(write: &mut W, feedback: &TracedEvent) -where +async fn send_interaction_feedback( + write: &mut W, + feedback: TracedEvent, + feedback_rx: &mut tokio::sync::mpsc::UnboundedReceiver, +) where W: WebsocketSinkExt, { - for op_id in operation_finished_ids(feedback) { - write - .send_json(&InteractionSessionMessage::Feedback( - InteractionSessionFeedback::OperationFinished(op_id), - )) - .await; + let mut events = Vec::new(); + append_interaction_feedback(&mut events, &feedback); + tokio::time::sleep(Duration::from_millis(1)).await; + for _ in 1..256 { + let Ok(feedback) = feedback_rx.try_recv() else { + break; + }; + append_interaction_feedback(&mut events, &feedback); } - - if let Some(op_id) = operation_started_id(feedback) { + if !events.is_empty() { write - .send_json(&InteractionSessionMessage::Feedback( - InteractionSessionFeedback::OperationStarted(op_id), - )) + .send_json(&InteractionSessionMessage::Feedback { events }) .await; } } #[cfg(feature = "router")] -fn operation_started_id(feedback: &TracedEvent) -> Option { - match &feedback.event { - TracedEventKind::MessageSent(message) => message - .input - .info - .as_ref() - .and_then(|info| info.id().as_ref()) - .map(ToString::to_string), - TracedEventKind::BufferEvent(event) => event - .accessor - .info - .as_ref() - .and_then(|info| info.id().as_ref()) - .map(ToString::to_string), - _ => None, - } +fn append_interaction_feedback( + events: &mut Vec, + feedback: &TracedEvent, +) { + events.extend(connection_activity_feedback(feedback)); + events.extend(operation_lifecycle_feedback(feedback)); +} + +#[cfg(feature = "router")] +fn interaction_operation_id(info: Option<&crossflow::OperationInfo>) -> Option { + info?.id().as_ref().map(ToString::to_string) +} + +#[cfg(feature = "router")] +fn operation_lifecycle_feedback(feedback: &TracedEvent) -> Option { + let (operation, started) = match &feedback.event { + TracedEventKind::OperationStarted(event) => (&event.operation, true), + TracedEventKind::OperationFinished(event) => (&event.operation, false), + _ => return None, + }; + let operation_id = interaction_operation_id(operation.info.as_deref())?; + let session = operation + .session_stack + .last() + .copied() + .unwrap_or(Entity::PLACEHOLDER) + .to_bits(); + let execution_id = format!("{session}:{}:{}", operation.target.to_bits(), operation.seq); + Some(if started { + InteractionSessionFeedback::OperationStarted { + operation_id, + execution_id, + } + } else { + InteractionSessionFeedback::OperationFinished { + operation_id, + execution_id, + } + }) } #[cfg(feature = "router")] -fn operation_finished_ids(feedback: &TracedEvent) -> Vec { +fn connection_activity_feedback(feedback: &TracedEvent) -> Vec { match &feedback.event { - TracedEventKind::MessageSent(message) => message - .output - .iter() - .filter_map(|source| { - source - .info - .as_ref() - .and_then(|info| info.id().as_ref()) - .map(ToString::to_string) - }) - .collect(), + TracedEventKind::MessageSent(message) => { + let Some(target_operation_id) = interaction_operation_id(message.input.info.as_deref()) + else { + return Vec::new(); + }; + message + .output + .iter() + .filter_map(|source| { + let source_operation_id = interaction_operation_id(source.info.as_deref())?; + Some(InteractionSessionFeedback::ConnectionActivity { + source_operation_id, + target_operation_id: target_operation_id.clone(), + }) + }) + .collect() + } + TracedEventKind::BufferEvent(event) => { + let Some(source_operation_id) = interaction_operation_id(event.buffer.info.as_deref()) + else { + return Vec::new(); + }; + let Some(target_operation_id) = + interaction_operation_id(event.accessor.info.as_deref()) + else { + return Vec::new(); + }; + if source_operation_id == target_operation_id { + return Vec::new(); + } + vec![InteractionSessionFeedback::ConnectionActivity { + source_operation_id, + target_operation_id, + }] + } _ => Vec::new(), } } @@ -1340,6 +1381,27 @@ mod tests { .unwrap() } + fn new_add7_chain_diagram(count: usize) -> Diagram { + let mut ops = serde_json::Map::new(); + for i in 0..count { + let next = if i + 1 == count { + json!({ "builtin": "terminate" }) + } else { + json!(format!("add7_{next}", next = i + 1)) + }; + ops.insert( + format!("add7_{i}"), + json!({ "type": "node", "builder": "add7", "next": next }), + ); + } + Diagram::from_json(json!({ + "version": "0.1.0", + "start": "add7_0", + "ops": ops, + })) + .unwrap() + } + #[tokio::test] #[test_log::test] async fn test_post_run() { @@ -1538,7 +1600,8 @@ mod tests { cleanup_test, } = setup_ws_test(); - let mut diagram = new_add7_diagram(); + const OPERATION_COUNT: usize = 1000; + let mut diagram = new_add7_chain_diagram(OPERATION_COUNT); diagram.default_trace = crossflow::TraceToggle::On; let request_body = PostRunRequest { @@ -1564,36 +1627,27 @@ mod tests { .await .unwrap(); - // There should be 4 feedback messages: add7 starts, add7 finishes, - // terminate starts, and terminate finishes. - for _ in 0..4 { + // Each operation and terminate start and finish, with one connection per operation. + let mut feedback_frames = 0; + let mut feedback_events = 0; + let resp = loop { let msg = test_rx.next().await.unwrap(); - let feedback_msg: InteractionSessionMessage = + let msg: serde_json::Value = serde_json::from_slice(msg.into_text().unwrap().as_bytes()).unwrap(); - let feedback = match feedback_msg { - InteractionSessionMessage::Feedback(feedback) => feedback, - _ => { - panic!("expected feedback message"); + match msg["type"].as_str() { + Some("feedback") => { + feedback_frames += 1; + feedback_events += msg["events"].as_array().map_or(1, Vec::len); } - }; - assert!(matches!( - feedback, - InteractionSessionFeedback::OperationStarted(_) - | InteractionSessionFeedback::OperationFinished(_) - )); - } - - let resp_msg = test_rx.next().await.unwrap(); - let resp_text = resp_msg.into_text().unwrap(); - let resp_msg: InteractionSessionMessage = - serde_json::from_slice(resp_text.as_bytes()).unwrap(); - let resp = match resp_msg { - InteractionSessionMessage::Finish(InteractionSessionEnd::Ok(resp)) => resp, - _ => { - panic!("expected response to be Ok"); + Some("finish") => { + break msg["ok"].clone(); + } + _ => panic!("unexpected interaction message"), } }; - assert_eq!(resp, serde_json::Value::from(12)); + assert_eq!(feedback_events, 3 * OPERATION_COUNT + 2); + assert!(feedback_frames * 10 < feedback_events); + assert_eq!(resp, serde_json::Value::from(5 + 7 * OPERATION_COUNT)); cleanup_test(); } diff --git a/src/diagram/buffer_schema.rs b/src/diagram/buffer_schema.rs index 0f5bf6f3..8958b8cc 100644 --- a/src/diagram/buffer_schema.rs +++ b/src/diagram/buffer_schema.rs @@ -420,10 +420,9 @@ pub struct ListenSchema { impl BuildDiagramOperation for ListenSchema { fn build_diagram_operation( &self, - _: &OperationName, + id: &OperationName, ctx: &mut BuilderContext, ) -> Result { - // TODO(@mxgrey): Figure out how to enable tracing for listen operations let target_type = ctx.inferred_message_type(&self.next)?; let buffer_map = match ctx.create_buffer_map(&self.buffers) { @@ -435,6 +434,8 @@ impl BuildDiagramOperation for ListenSchema { .registry .messages .listen(&target_type, &buffer_map, ctx.builder)?; + let trace = TraceInfo::new(self, self.trace_settings.trace)?; + ctx.trace_output_source(id, &output, trace); ctx.add_output_into_target(&self.next, output); Ok(BuildStatus::Finished) } diff --git a/src/diagram/join_schema.rs b/src/diagram/join_schema.rs index 061498ca..fcf07c5f 100644 --- a/src/diagram/join_schema.rs +++ b/src/diagram/join_schema.rs @@ -21,7 +21,7 @@ use std::borrow::Cow; use super::{ BufferSelection, BuildDiagramOperation, BuildStatus, BuilderContext, DiagramErrorCode, - JsonMessage, NextOperation, OperationName, + JsonMessage, NextOperation, OperationName, TraceInfo, }; use crate::{ BufferMap, BufferMapLayout, BufferMapLayoutHints, Builder, DynOutput, IdentifierRef, @@ -133,18 +133,21 @@ impl BuildDiagramOperation for JoinSchema { })?; } - if self.serialize { - let output = ctx.builder.try_join::(&buffer_map)?.output(); - ctx.add_output_into_target(&self.next, output.into()); + let output: DynOutput = if self.serialize { + ctx.builder + .try_join::(&buffer_map)? + .output() + .into() } else { let target_type = ctx.inferred_message_type(output_ref(id).next())?; - let output = ctx - .registry + ctx.registry .messages - .join(&target_type, &buffer_map, ctx.builder)?; - ctx.add_output_into_target(&self.next, output); - } + .join(&target_type, &buffer_map, ctx.builder)? + }; + let trace = TraceInfo::new(self, self.trace_settings.trace)?; + ctx.trace_output_source(id, &output, trace); + ctx.add_output_into_target(&self.next, output); Ok(BuildStatus::Finished) } diff --git a/src/diagram/workflow_builder.rs b/src/diagram/workflow_builder.rs index 360bc250..718d4954 100644 --- a/src/diagram/workflow_builder.rs +++ b/src/diagram/workflow_builder.rs @@ -131,6 +131,40 @@ impl<'a, 'c, 'w, 's, 'b> BuilderContext<'a, 'c, 'w, 's, 'b> { .push(output); } + /// Attach tracing to an operation whose input is driven internally. + pub fn trace_output_source( + &mut self, + #[allow(unused)] operation: impl Into, + #[allow(unused)] output: &DynOutput, + #[allow(unused)] trace_info: TraceInfo, + ) { + #[cfg(feature = "trace")] + { + let operation = self.into_operation_ref(operation); + let operation_info = OperationInfo::new( + Some(operation), + Some(output.message_info().type_name.into()), + trace_info.construction, + ); + let trace = Trace::new( + trace_info.trace.unwrap_or(self.default_trace), + Arc::new(operation_info), + ); + let output = output.id(); + self.builder + .commands() + .queue(move |world: &mut bevy_ecs::world::World| { + let source = world + .get::(output) + .and_then(|inputs| inputs.get().first()) + .copied(); + if let Some(source) = source { + world.entity_mut(source).insert(trace); + } + }); + } + } + /// Set the input slot of an operation. This should not be called more than /// once per operation, because only one input slot can be used for any /// operation. diff --git a/src/input.rs b/src/input.rs index 9627c19f..333077e7 100644 --- a/src/input.rs +++ b/src/input.rs @@ -37,6 +37,7 @@ use crate::{ #[cfg(feature = "trace")] use crate::{ Debug, DebugRoster, MessageSent, Trace, TraceToggle, TracedEvent, UniversalTraceToggle, + is_traced_request, trace_operation_started, }; pub type Seq = u32; @@ -407,11 +408,13 @@ impl ManageInput for World { if !perform_trace { // Check if any of the sources want to trace for output in &route.outputs { - if let Some(trace) = self.get::(output.source) { - if trace.toggle().is_on() { - perform_trace = true; - break; - } + if self + .get::(output.source) + .is_some_and(|trace| trace.toggle().is_on()) + || is_traced_request(output.request_id(), self) + { + perform_trace = true; + break; } } } @@ -498,7 +501,7 @@ impl ManageInput for World { #[cfg(feature = "trace")] { self.get_resource_or_init::(); - self.resource_scope::(|world, mut debug| { + let input = self.resource_scope::(|world, mut debug| { if !debug.is_active() { // Revert to the usual implementation of popping the next let mut storage = world.get_mut::>(source).or_broken()?; @@ -546,7 +549,18 @@ impl ManageInput for World { } }) } - }) + })?; + if let Some(input) = &input { + trace_operation_started( + RequestId { + session: input.session, + source, + seq: input.seq, + }, + self, + ); + } + Ok(input) } } diff --git a/src/operation.rs b/src/operation.rs index 3439c7a1..39ff11a1 100644 --- a/src/operation.rs +++ b/src/operation.rs @@ -774,6 +774,9 @@ fn perform_operation( world.emit_broken(source, backtrace, roster); } } + + #[cfg(feature = "trace")] + crate::trace_immediate_operations_finished(source, world); } pub struct DownstreamIter<'a> { diff --git a/src/operation/operate_task.rs b/src/operation/operate_task.rs index c65bd7eb..be817677 100644 --- a/src/operation/operate_task.rs +++ b/src/operation/operate_task.rs @@ -109,6 +109,8 @@ impl OperateTask(self.node()).map(|s| s.scope()); let mut source_mut = world.entity_mut(source); source_mut.insert(ChildOf(self.node())); @@ -154,6 +156,8 @@ where sender .send(Box::new( move |world: &mut World, roster: &mut OperationRoster| { + #[cfg(feature = "trace")] + crate::trace_operation_finished(request_id, world); cleanup_task(source, node, unblock, being_cleaned, world, roster); if disposed { @@ -262,6 +266,8 @@ where .get_mut::>(source) .or_broken()? .finished_normally = true; + #[cfg(feature = "trace")] + crate::trace_operation_finished(request_id, world); cleanup_task(source, node, unblock, being_cleaned, world, roster); if Streams::has_streams() { @@ -338,6 +344,8 @@ where .or_broken()?; operation.being_cleaned = Some(cleanup); operation.finished_normally = true; + #[cfg(feature = "trace")] + let request_id = operation.request_id; let node = operation.node(); let task = operation.task.take(); let unblock = operation.blocker.take(); @@ -347,6 +355,8 @@ where task.cancel().await; if let Err(err) = sender.send(Box::new( move |world: &mut World, roster: &mut OperationRoster| { + #[cfg(feature = "trace")] + crate::trace_operation_finished(request_id, world); cleanup_task(source, node, unblock, Some(cleanup), world, roster); }, )) { @@ -354,6 +364,8 @@ where } }); } else { + #[cfg(feature = "trace")] + crate::trace_operation_finished(request_id, clean.world); cleanup_task( source, node, diff --git a/src/trace.rs b/src/trace.rs index 6fabf722..53f72b29 100644 --- a/src/trace.rs +++ b/src/trace.rs @@ -32,7 +32,7 @@ use smallvec::SmallVec; use std::{ any::Any, borrow::Cow, - collections::VecDeque, + collections::{HashMap, HashSet, VecDeque}, sync::Arc, time::{Instant, SystemTime}, }; @@ -179,24 +179,30 @@ pub struct TraceSource { impl TraceSource { fn new(route_source: RouteSource, world: &mut World) -> Self { + let trace_owner = find_trace_owner(route_source.source, world); + let request_info = world + .get_resource::() + .and_then(|tracker| tracker.operations.get(&route_source.request_id())) + .cloned(); let output_port = route_source.port; let session_stack = get_session_stack_from_world(route_source.session, world); let port = route_source.port.iter().map(|p| p.to_owned()).collect(); - let operation_type = world - .get::(route_source.source) + let operation_type = trace_owner + .and_then(|entity| world.get::(entity)) .map(|op| (**op).clone()) .unwrap_or_else(|| "".into()); - let info = world - .get::(route_source.source) - .map(|t| t.info.clone()); - + let info = request_info.or_else(|| { + trace_owner + .and_then(|entity| world.get::(entity)) + .map(|trace| trace.info.clone()) + }); TraceSource { session_stack, source: route_source.source, seq: route_source.seq, port, - labels: world - .get::(route_source.source) + labels: trace_owner + .and_then(|entity| world.get::(entity)) .map(move |labels| labels.outputs(output_port)) .flatten(), operation_type, @@ -205,6 +211,15 @@ impl TraceSource { } } +fn find_trace_owner(mut entity: Entity, world: &World) -> Option { + loop { + if world.get::(entity).is_some() { + return Some(entity); + } + entity = world.get::(entity)?.parent(); + } +} + #[derive(Debug, Clone)] pub struct TraceTarget { /// The stack of session IDs that sent the message was sent into. The first @@ -267,6 +282,7 @@ pub struct TraceBuffer { /// The unique ID of the buffer. pub id: Entity, pub labels: Option>>, + pub info: Option>, } pub type TracedMessage = Option>; @@ -293,19 +309,27 @@ impl MessageSent { message: TracedMessage, world: &mut World, ) { - let mut output = SmallVec::new(); + let mut output: SmallVec<[TraceSource; 8]> = SmallVec::new(); for out in route.outputs { output.push(TraceSource::new(out, world)); } - let input = TraceTarget::new( - RequestId { - session: route.input.session, - source: route.input.target, - seq: target_seq, - }, - world, - ); + let target_request = RequestId { + session: route.input.session, + source: route.input.target, + seq: target_seq, + }; + let input = TraceTarget::new(target_request, world); + let request_info = input + .info + .clone() + .or_else(|| output.iter().find_map(|source| source.info.clone())); + if let Some(info) = request_info { + world + .get_resource_or_init::() + .operations + .insert(target_request, info); + } let event = MessageSent { output, @@ -316,6 +340,86 @@ impl MessageSent { } } +pub(crate) fn is_traced_request(request_id: RequestId, world: &World) -> bool { + world + .get_resource::() + .is_some_and(|tracker| tracker.operations.contains_key(&request_id)) +} + +/// Tracks the lifecycle of one invocation of an operation. +#[derive(Debug, Clone)] +pub struct OperationLifecycle { + pub operation: TraceTarget, +} + +#[derive(Resource, Default)] +struct OperationLifecycleTracker { + active: HashSet, + deferred: HashSet, + operations: HashMap>, +} + +pub(crate) fn trace_operation_started(request_id: RequestId, world: &mut World) { + let universal = world + .get_resource::() + .and_then(|toggle| **toggle); + let trace_owner = find_trace_owner(request_id.source, world); + let toggle = universal + .or_else(|| trace_owner.and_then(|entity| world.get::(entity).map(Trace::toggle))); + if !toggle.is_some_and(|toggle| toggle.is_on()) && !is_traced_request(request_id, world) { + return; + } + + let operation = TraceTarget::new(request_id, world); + let mut tracker = world.get_resource_or_init::(); + tracker.active.insert(request_id); + if let Some(info) = &operation.info { + tracker.operations.insert(request_id, Arc::clone(info)); + } + world.write_trace(TracedEvent::now(TracedEventKind::OperationStarted( + OperationLifecycle { operation }, + ))); +} + +pub(crate) fn defer_operation_finished(request_id: RequestId, world: &mut World) { + let mut tracker = world.get_resource_or_init::(); + if tracker.active.contains(&request_id) { + tracker.deferred.insert(request_id); + } +} + +pub(crate) fn trace_operation_finished(request_id: RequestId, world: &mut World) { + let should_trace = { + let mut tracker = world.get_resource_or_init::(); + tracker.deferred.remove(&request_id); + tracker.operations.remove(&request_id); + tracker.active.remove(&request_id) + }; + if !should_trace { + return; + } + + let operation = TraceTarget::new(request_id, world); + world.write_trace(TracedEvent::now(TracedEventKind::OperationFinished( + OperationLifecycle { operation }, + ))); +} + +pub(crate) fn trace_immediate_operations_finished(source: Entity, world: &mut World) { + let ready = { + let tracker = world.get_resource_or_init::(); + tracker + .active + .iter() + .filter(|request| request.source == source && !tracker.deferred.contains(request)) + .copied() + .collect::>() + }; + for request_id in ready { + trace_operation_finished(request_id, world); + } +} + /// Track which outputs of an operation did not yield any message after the /// operation was triggered. #[derive(Debug, Clone)] @@ -533,10 +637,12 @@ impl<'w, 's> BufferTracer<'w, 's> { pub(crate) fn get_trace_buffer(&self, key: &BufferKeyTag) -> TraceBuffer { let session_stack = get_session_stack(key.session, &self.child_of); let labels = self.labels.get(key.buffer).ok().map(|l| l.input.clone()); + let info = self.trace.get(key.buffer).ok().map(|t| t.info.clone()); TraceBuffer { session_stack, id: key.buffer, labels: labels, + info, } } @@ -585,6 +691,15 @@ impl SessionEvent { pub(crate) fn despawned(session: Entity, world: &mut World) { let session_stack = get_session_stack_from_world(session, world); + if let Some(mut tracker) = world.get_resource_mut::() { + tracker.active.retain(|request| request.session != session); + tracker + .deferred + .retain(|request| request.session != session); + tracker + .operations + .retain(|request, _| request.session != session); + } let event = SessionEvent { session_stack, change: SessionChange::Despawned, @@ -683,6 +798,10 @@ pub enum PauseCause { pub enum TracedEventKind { /// A message was sent from one operation to another MessageSent(MessageSent), + /// An operation began processing one request. + OperationStarted(OperationLifecycle), + /// An operation finished processing one request. + OperationFinished(OperationLifecycle), /// A buffer was viewed or modified by an operation BufferEvent(BufferEvent), /// A session was spawned despawned, or changed state @@ -708,6 +827,9 @@ impl TracedEventKind { } } } + Self::OperationStarted(event) | Self::OperationFinished(event) => { + return event.operation.session_stack.contains(&session); + } Self::SessionEvent(s) => { return s.session_stack.contains(&session); }