From 1ed53f7ae78b6f76b8b11d78a67acb62229e890f Mon Sep 17 00:00:00 2001 From: shrimbly Date: Thu, 5 Feb 2026 00:16:38 +1300 Subject: [PATCH] fix: resolve resource leaks, stale state refs, and cache correctness issues - Split AudioInputNode waveform effect to avoid ResizeObserver churn on every animation tick - Add missing kieApiKey/wavespeedApiKey deps to ModelParameters useEffect - Clean up VideoSampleSource/Output on error in useApplySpeedCurve - Fix off-by-one in useStitchVideos duplicate-frame skip (<= to <) and add try/finally for resource cleanup - Replace leftover currentNodeId: null with currentNodeIds: [] in workflowStore - Store full response metadata (status, headers, body) in deduplicatedFetch cache - Add in-flight guard to imageStorage to prevent duplicate concurrent saves Co-Authored-By: Claude Opus 4.5 --- src/components/nodes/AudioInputNode.tsx | 91 ++++++++++++++---------- src/components/nodes/ModelParameters.tsx | 2 +- src/hooks/useApplySpeedCurve.ts | 27 ++++++- src/hooks/useStitchVideos.ts | 47 ++++++++++-- src/store/workflowStore.ts | 10 +-- src/utils/deduplicatedFetch.ts | 39 +++++----- src/utils/imageStorage.ts | 56 ++++++++++----- 7 files changed, 185 insertions(+), 87 deletions(-) diff --git a/src/components/nodes/AudioInputNode.tsx b/src/components/nodes/AudioInputNode.tsx index 951fdfb9..f6f4d7b0 100644 --- a/src/components/nodes/AudioInputNode.tsx +++ b/src/components/nodes/AudioInputNode.tsx @@ -63,7 +63,31 @@ export function AudioInputNode({ id, data, selected }: NodeProps { + ctx.clearRect(0, 0, width, height); + + const barCount = Math.min(peaks.length, width); + const barWidth = width / barCount; + const barGap = 1; + + ctx.fillStyle = "rgb(167, 139, 250)"; // violet-400 + + for (let i = 0; i < barCount; i++) { + const peakIndex = Math.floor((i / barCount) * peaks.length); + const peak = peaks[peakIndex] || 0; + const barHeight = peak * height; + const x = i * barWidth; + const y = (height - barHeight) / 2; + + ctx.fillRect(x, y, barWidth - barGap, barHeight); + } + }, + [] + ); + + // Effect A: ResizeObserver — only recreated when waveformData changes useEffect(() => { if (!waveformData || !canvasRef.current || !waveformContainerRef.current) return; @@ -72,48 +96,15 @@ export function AudioInputNode({ id, data, selected }: NodeProps { for (const entry of entries) { const width = entry.contentRect.width; const height = entry.contentRect.height; - // Set canvas size canvas.width = width; canvas.height = height; - // Draw waveform - ctx.clearRect(0, 0, width, height); - - const peaks = waveformData.peaks; - const barCount = Math.min(peaks.length, width); - const barWidth = width / barCount; - const barGap = 1; - - ctx.fillStyle = "rgb(167, 139, 250)"; // violet-400 - - for (let i = 0; i < barCount; i++) { - const peakIndex = Math.floor((i / barCount) * peaks.length); - const peak = peaks[peakIndex] || 0; - const barHeight = peak * height; - const x = i * barWidth; - const y = (height - barHeight) / 2; - - ctx.fillRect(x, y, barWidth - barGap, barHeight); - } - - // Draw playback position - if (isPlaying && nodeData.duration) { - const progress = currentTime / nodeData.duration; - const x = progress * width; - - ctx.strokeStyle = "rgba(255, 255, 255, 0.8)"; - ctx.lineWidth = 2; - ctx.beginPath(); - ctx.moveTo(x, 0); - ctx.lineTo(x, height); - ctx.stroke(); - } + drawWaveform(ctx, width, height, waveformData.peaks); } }); @@ -122,7 +113,35 @@ export function AudioInputNode({ id, data, selected }: NodeProps { resizeObserver.disconnect(); }; - }, [waveformData, isPlaying, currentTime, nodeData.duration]); + }, [waveformData, drawWaveform]); + + // Effect B: Redraw waveform + playback position (lightweight, no ResizeObserver) + useEffect(() => { + if (!waveformData || !canvasRef.current) return; + + const canvas = canvasRef.current; + const ctx = canvas.getContext("2d"); + if (!ctx) return; + + const width = canvas.width; + const height = canvas.height; + if (width === 0 || height === 0) return; + + drawWaveform(ctx, width, height, waveformData.peaks); + + // Draw playback position + if (isPlaying && nodeData.duration) { + const progress = currentTime / nodeData.duration; + const x = progress * width; + + ctx.strokeStyle = "rgba(255, 255, 255, 0.8)"; + ctx.lineWidth = 2; + ctx.beginPath(); + ctx.moveTo(x, 0); + ctx.lineTo(x, height); + ctx.stroke(); + } + }, [isPlaying, currentTime, nodeData.duration, waveformData, drawWaveform]); // Animation loop for smooth playback position updates useEffect(() => { diff --git a/src/components/nodes/ModelParameters.tsx b/src/components/nodes/ModelParameters.tsx index 91cbf7ef..4b71292f 100644 --- a/src/components/nodes/ModelParameters.tsx +++ b/src/components/nodes/ModelParameters.tsx @@ -91,7 +91,7 @@ export function ModelParameters({ }; fetchSchema(); - }, [modelId, provider, replicateApiKey, falApiKey, onInputsLoaded]); + }, [modelId, provider, replicateApiKey, falApiKey, kieApiKey, wavespeedApiKey, onInputsLoaded]); // Notify parent to resize node when schema loads and panel is expanded useEffect(() => { diff --git a/src/hooks/useApplySpeedCurve.ts b/src/hooks/useApplySpeedCurve.ts index b2d7b82a..6ec7268a 100644 --- a/src/hooks/useApplySpeedCurve.ts +++ b/src/hooks/useApplySpeedCurve.ts @@ -80,6 +80,9 @@ async function applySpeedCurveCore( bitrate: number ): Promise { let input: Input | null = null; + let videoSource: VideoSampleSource | null = null; + let output: Output | null = null; + let outputStarted = false; try { // Helper to update progress @@ -250,7 +253,7 @@ async function applySpeedCurveCore( 22 ); - const videoSource = new VideoSampleSource( + videoSource = new VideoSampleSource( createAvcEncodingConfig( selectedConfig.bitrate, selectedConfig.width, @@ -261,7 +264,7 @@ async function applySpeedCurveCore( ); const bufferTarget = new BufferTarget(); - const output = new Output({ + output = new Output({ format: new Mp4OutputFormat({ fastStart: 'in-memory' }), target: bufferTarget, }); @@ -271,6 +274,7 @@ async function applySpeedCurveCore( updateProgress('processing', 'Starting output encoding...', 25); await output.start(); + outputStarted = true; // Step 4: OUTPUT-DRIVEN FRAME EMISSION const minFrameInterval = 1 / selectedConfig.framerate; @@ -289,7 +293,7 @@ async function applySpeedCurveCore( const outputSample = (sourceSample as VideoSample).clone(); outputSample.setTimestamp(timestamp); outputSample.setDuration(duration); - await videoSource.add(outputSample); + await videoSource!.add(outputSample); outputSample.close(); }; @@ -359,8 +363,11 @@ async function applySpeedCurveCore( // Ensure encoder flushes SPS/PPS before finalizing await videoSource.close(); + videoSource = null; // Already closed, prevent double-close in finally // Step 5: Finalize and get output blob await output.finalize(); + outputStarted = false; // Successfully finalized, no need to cancel in finally + output = null; const buffer = bufferTarget.buffer; if (!buffer) { @@ -377,6 +384,20 @@ async function applySpeedCurveCore( return outputBlob; } finally { + if (videoSource) { + try { + await videoSource.close(); + } catch (e) { + console.warn('Failed to close videoSource:', e); + } + } + if (output && outputStarted) { + try { + await output.cancel(); + } catch (e) { + console.warn('Failed to cancel output:', e); + } + } if (input) { try { input.dispose(); diff --git a/src/hooks/useStitchVideos.ts b/src/hooks/useStitchVideos.ts index fe405ff0..26cb4eb5 100644 --- a/src/hooks/useStitchVideos.ts +++ b/src/hooks/useStitchVideos.ts @@ -265,12 +265,12 @@ export async function stitchVideosAsync( // Create output updateProgress('processing', 'Creating output container...', 10); - const videoSource = new VideoSampleSource( + let videoSource: VideoSampleSource | null = new VideoSampleSource( createAvcEncodingConfig(resolvedBitrate, safeWidth, safeHeight, codecProfile) ); const bufferTarget = new BufferTarget(); - const output = new Output({ + let output: Output | null = new Output({ format: new Mp4OutputFormat({ fastStart: 'in-memory' }), target: bufferTarget, }); @@ -278,8 +278,10 @@ export async function stitchVideosAsync( output.addVideoTrack(videoSource, { rotation: aggregateRotation, frameRate: MAX_OUTPUT_FPS }); // Add audio track if provided - let audioSource: AudioBufferSource | undefined; + let audioSource: AudioBufferSource | null = null; let pendingAudioBuffer: AudioBuffer | null = null; + let outputStarted = false; + if (audioData) { updateProgress('processing', 'Detecting supported audio codec...', 8); @@ -308,7 +310,9 @@ export async function stitchVideosAsync( } await output.start(); + outputStarted = true; + try { // Track the highest timestamp we've written to ensure monotonicity // Start at -frameInterval so first frame can be at timestamp 0 const frameInterval = 1 / MAX_OUTPUT_FPS; @@ -370,14 +374,14 @@ export async function stitchVideosAsync( // Skip duplicate frames that land on the same timestamp slot // This ensures strict 60fps without exceeding the target rate - if (snappedTimestamp <= highestWrittenTimestamp) { + if (snappedTimestamp < highestWrittenTimestamp) { sample.close(); continue; } sample.setTimestamp(snappedTimestamp); sample.setDuration(frameInterval); - await videoSource.add(sample); + await videoSource!.add(sample); // Update highest written timestamp highestWrittenTimestamp = snappedTimestamp; @@ -418,14 +422,18 @@ export async function stitchVideosAsync( const trimmedBuffer = trimAudioBuffer(pendingAudioBuffer, videoDuration); await audioSource.add(trimmedBuffer); await audioSource.close(); + audioSource = null; // Already closed } // Flush encoder before finalizing container - await videoSource.close(); + await videoSource!.close(); + videoSource = null; // Already closed, prevent double-close in finally updateProgress('processing', 'Finalizing stitched video...', 97); // Finalize output - await output.finalize(); + await output!.finalize(); + outputStarted = false; // Successfully finalized + output = null; const buffer = bufferTarget.buffer; if (!buffer) { @@ -441,6 +449,31 @@ export async function stitchVideosAsync( ); return outputBlob; + } finally { + // Clean up encoding resources on error + pendingAudioBuffer = null; + if (audioSource) { + try { + await audioSource.close(); + } catch (e) { + console.warn('Failed to close audioSource:', e); + } + } + if (videoSource) { + try { + await videoSource.close(); + } catch (e) { + console.warn('Failed to close videoSource:', e); + } + } + if (output && outputStarted) { + try { + await output.cancel(); + } catch (e) { + console.warn('Failed to cancel output:', e); + } + } + } } catch (error) { const normalizedError = error instanceof Error ? error : new Error(String(error)); diff --git a/src/store/workflowStore.ts b/src/store/workflowStore.ts index b3b09093..84e48143 100644 --- a/src/store/workflowStore.ts +++ b/src/store/workflowStore.ts @@ -2938,7 +2938,7 @@ export const useWorkflowStore = create((set, get) => ({ error: "Browser does not support video encoding", progress: 0, }); - set({ isRunning: false, currentNodeId: null }); + set({ isRunning: false, currentNodeIds: [] }); await logger.endSession(); return; } @@ -2954,7 +2954,7 @@ export const useWorkflowStore = create((set, get) => ({ error: "Need at least 2 video clips to stitch", progress: 0, }); - set({ isRunning: false, currentNodeId: null }); + set({ isRunning: false, currentNodeIds: [] }); await logger.endSession(); return; } @@ -3018,7 +3018,7 @@ export const useWorkflowStore = create((set, get) => ({ error: "Browser does not support video encoding", progress: 0, }); - set({ isRunning: false, currentNodeId: null }); + set({ isRunning: false, currentNodeIds: [] }); await logger.endSession(); return; } @@ -3051,7 +3051,7 @@ export const useWorkflowStore = create((set, get) => ({ error: "Connect a video input to apply ease curve", progress: 0, }); - set({ isRunning: false, currentNodeId: null }); + set({ isRunning: false, currentNodeIds: [] }); await logger.endSession(); return; } @@ -3127,7 +3127,7 @@ export const useWorkflowStore = create((set, get) => ({ }); } - set({ isRunning: false, currentNodeId: null }); + set({ isRunning: false, currentNodeIds: [] }); await logger.endSession(); return; } diff --git a/src/utils/deduplicatedFetch.ts b/src/utils/deduplicatedFetch.ts index aab8b641..c5f70995 100644 --- a/src/utils/deduplicatedFetch.ts +++ b/src/utils/deduplicatedFetch.ts @@ -13,8 +13,8 @@ // Map of URL -> in-flight promise const inFlightRequests = new Map>(); -// Map of URL -> cloned response data (since Response body can only be read once) -const responseCache = new Map(); +// Map of URL -> cloned response metadata (since Response body can only be read once) +const responseCache = new Map; bodyText: string; timestamp: number }>(); // Cache TTL in milliseconds (5 seconds - short enough to get fresh data, long enough to dedupe) const CACHE_TTL = 5000; @@ -59,10 +59,10 @@ export async function deduplicatedFetch( // Check if we have a recent cached response const cached = responseCache.get(cacheKey); if (cached && Date.now() - cached.timestamp < CACHE_TTL) { - // Return a synthetic response with the cached data - return new Response(JSON.stringify(cached.data), { - status: 200, - headers: { "Content-Type": "application/json" }, + // Return a synthetic response with the cached metadata + return new Response(cached.bodyText, { + status: cached.status, + headers: cached.headers, }); } @@ -70,13 +70,13 @@ export async function deduplicatedFetch( const existingRequest = inFlightRequests.get(cacheKey); if (existingRequest) { // Wait for the existing request and clone it for this caller - const response = await existingRequest; + await existingRequest; // The response body may have been consumed, so we rely on the cache const cachedData = responseCache.get(cacheKey); if (cachedData) { - return new Response(JSON.stringify(cachedData.data), { - status: response.status, - headers: { "Content-Type": "application/json" }, + return new Response(cachedData.bodyText, { + status: cachedData.status, + headers: cachedData.headers, }); } // Fallback: this shouldn't happen, but return an error response @@ -88,14 +88,17 @@ export async function deduplicatedFetch( // Create new request const requestPromise = fetch(url, options) .then(async (response) => { - // Clone and cache the response data before returning - if (response.ok) { - try { - const data = await response.clone().json(); - responseCache.set(cacheKey, { data, timestamp: Date.now() }); - } catch { - // Response wasn't JSON, that's fine - } + // Clone and cache the full response metadata before returning + try { + const bodyText = await response.clone().text(); + responseCache.set(cacheKey, { + status: response.status, + headers: Object.fromEntries(response.headers.entries()), + bodyText, + timestamp: Date.now(), + }); + } catch { + // Failed to read response body for caching } return response; }) diff --git a/src/utils/imageStorage.ts b/src/utils/imageStorage.ts index b7f2799c..0a3fa6b8 100644 --- a/src/utils/imageStorage.ts +++ b/src/utils/imageStorage.ts @@ -295,6 +295,9 @@ async function externalizeNodeImages( } as WorkflowNode; } +// In-flight saves guard to prevent duplicate concurrent uploads of the same image +const inFlightSaves = new Map>(); + /** * Save an image and return its ID (with deduplication) * @param folder - "inputs" for user-uploaded images, "generations" for AI-generated images @@ -317,31 +320,50 @@ async function saveImageAndGetId( return savedImageIds.get(hash)!; } + // Check if there's already an in-flight save for this hash + if (!existingId && inFlightSaves.has(hash)) { + return inFlightSaves.get(hash)!; + } + // Use existing ID if provided (for consistency with imageHistory), otherwise generate new const imageId = existingId || generateImageId(); - const response = await fetchWithTimeout( - "/api/workflow-images", - { - method: "POST", - headers: { "Content-Type": "application/json" }, - body: JSON.stringify({ - workflowPath, - imageId, - imageData, - folder, - }), + const savePromise = (async () => { + const response = await fetchWithTimeout( + "/api/workflow-images", + { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ + workflowPath, + imageId, + imageData, + folder, + }), + } + ); + + const result = await response.json(); + + if (!result.success) { + throw new Error(`Failed to save image: ${result.error}`); } - ); - const result = await response.json(); + savedImageIds.set(hash, imageId); + return imageId; + })(); - if (!result.success) { - throw new Error(`Failed to save image: ${result.error}`); + if (!existingId) { + inFlightSaves.set(hash, savePromise); } - savedImageIds.set(hash, imageId); - return imageId; + try { + return await savePromise; + } catch (error) { + throw error; + } finally { + inFlightSaves.delete(hash); + } } /**