Browse Source

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 <noreply@anthropic.com>
handoff-20260429-1057
shrimbly 5 months ago
parent
commit
1ed53f7ae7
  1. 91
      src/components/nodes/AudioInputNode.tsx
  2. 2
      src/components/nodes/ModelParameters.tsx
  3. 27
      src/hooks/useApplySpeedCurve.ts
  4. 47
      src/hooks/useStitchVideos.ts
  5. 10
      src/store/workflowStore.ts
  6. 39
      src/utils/deduplicatedFetch.ts
  7. 56
      src/utils/imageStorage.ts

91
src/components/nodes/AudioInputNode.tsx

@ -63,7 +63,31 @@ export function AudioInputNode({ id, data, selected }: NodeProps<AudioInputNodeT
}; };
}, [nodeData.audioFile]); }, [nodeData.audioFile]);
// Draw waveform on canvas // Helper to draw waveform bars on canvas
const drawWaveform = useCallback(
(ctx: CanvasRenderingContext2D, width: number, height: number, peaks: number[]) => {
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(() => { useEffect(() => {
if (!waveformData || !canvasRef.current || !waveformContainerRef.current) return; if (!waveformData || !canvasRef.current || !waveformContainerRef.current) return;
@ -72,48 +96,15 @@ export function AudioInputNode({ id, data, selected }: NodeProps<AudioInputNodeT
const ctx = canvas.getContext("2d"); const ctx = canvas.getContext("2d");
if (!ctx) return; if (!ctx) return;
// Use ResizeObserver for responsive sizing
const resizeObserver = new ResizeObserver((entries) => { const resizeObserver = new ResizeObserver((entries) => {
for (const entry of entries) { for (const entry of entries) {
const width = entry.contentRect.width; const width = entry.contentRect.width;
const height = entry.contentRect.height; const height = entry.contentRect.height;
// Set canvas size
canvas.width = width; canvas.width = width;
canvas.height = height; canvas.height = height;
// Draw waveform drawWaveform(ctx, width, height, waveformData.peaks);
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();
}
} }
}); });
@ -122,7 +113,35 @@ export function AudioInputNode({ id, data, selected }: NodeProps<AudioInputNodeT
return () => { return () => {
resizeObserver.disconnect(); 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 // Animation loop for smooth playback position updates
useEffect(() => { useEffect(() => {

2
src/components/nodes/ModelParameters.tsx

@ -91,7 +91,7 @@ export function ModelParameters({
}; };
fetchSchema(); 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 // Notify parent to resize node when schema loads and panel is expanded
useEffect(() => { useEffect(() => {

27
src/hooks/useApplySpeedCurve.ts

@ -80,6 +80,9 @@ async function applySpeedCurveCore(
bitrate: number bitrate: number
): Promise<Blob> { ): Promise<Blob> {
let input: Input | null = null; let input: Input | null = null;
let videoSource: VideoSampleSource | null = null;
let output: Output | null = null;
let outputStarted = false;
try { try {
// Helper to update progress // Helper to update progress
@ -250,7 +253,7 @@ async function applySpeedCurveCore(
22 22
); );
const videoSource = new VideoSampleSource( videoSource = new VideoSampleSource(
createAvcEncodingConfig( createAvcEncodingConfig(
selectedConfig.bitrate, selectedConfig.bitrate,
selectedConfig.width, selectedConfig.width,
@ -261,7 +264,7 @@ async function applySpeedCurveCore(
); );
const bufferTarget = new BufferTarget(); const bufferTarget = new BufferTarget();
const output = new Output({ output = new Output({
format: new Mp4OutputFormat({ fastStart: 'in-memory' }), format: new Mp4OutputFormat({ fastStart: 'in-memory' }),
target: bufferTarget, target: bufferTarget,
}); });
@ -271,6 +274,7 @@ async function applySpeedCurveCore(
updateProgress('processing', 'Starting output encoding...', 25); updateProgress('processing', 'Starting output encoding...', 25);
await output.start(); await output.start();
outputStarted = true;
// Step 4: OUTPUT-DRIVEN FRAME EMISSION // Step 4: OUTPUT-DRIVEN FRAME EMISSION
const minFrameInterval = 1 / selectedConfig.framerate; const minFrameInterval = 1 / selectedConfig.framerate;
@ -289,7 +293,7 @@ async function applySpeedCurveCore(
const outputSample = (sourceSample as VideoSample).clone(); const outputSample = (sourceSample as VideoSample).clone();
outputSample.setTimestamp(timestamp); outputSample.setTimestamp(timestamp);
outputSample.setDuration(duration); outputSample.setDuration(duration);
await videoSource.add(outputSample); await videoSource!.add(outputSample);
outputSample.close(); outputSample.close();
}; };
@ -359,8 +363,11 @@ async function applySpeedCurveCore(
// Ensure encoder flushes SPS/PPS before finalizing // Ensure encoder flushes SPS/PPS before finalizing
await videoSource.close(); await videoSource.close();
videoSource = null; // Already closed, prevent double-close in finally
// Step 5: Finalize and get output blob // Step 5: Finalize and get output blob
await output.finalize(); await output.finalize();
outputStarted = false; // Successfully finalized, no need to cancel in finally
output = null;
const buffer = bufferTarget.buffer; const buffer = bufferTarget.buffer;
if (!buffer) { if (!buffer) {
@ -377,6 +384,20 @@ async function applySpeedCurveCore(
return outputBlob; return outputBlob;
} finally { } 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) { if (input) {
try { try {
input.dispose(); input.dispose();

47
src/hooks/useStitchVideos.ts

@ -265,12 +265,12 @@ export async function stitchVideosAsync(
// Create output // Create output
updateProgress('processing', 'Creating output container...', 10); updateProgress('processing', 'Creating output container...', 10);
const videoSource = new VideoSampleSource( let videoSource: VideoSampleSource | null = new VideoSampleSource(
createAvcEncodingConfig(resolvedBitrate, safeWidth, safeHeight, codecProfile) createAvcEncodingConfig(resolvedBitrate, safeWidth, safeHeight, codecProfile)
); );
const bufferTarget = new BufferTarget(); const bufferTarget = new BufferTarget();
const output = new Output({ let output: Output | null = new Output({
format: new Mp4OutputFormat({ fastStart: 'in-memory' }), format: new Mp4OutputFormat({ fastStart: 'in-memory' }),
target: bufferTarget, target: bufferTarget,
}); });
@ -278,8 +278,10 @@ export async function stitchVideosAsync(
output.addVideoTrack(videoSource, { rotation: aggregateRotation, frameRate: MAX_OUTPUT_FPS }); output.addVideoTrack(videoSource, { rotation: aggregateRotation, frameRate: MAX_OUTPUT_FPS });
// Add audio track if provided // Add audio track if provided
let audioSource: AudioBufferSource | undefined; let audioSource: AudioBufferSource | null = null;
let pendingAudioBuffer: AudioBuffer | null = null; let pendingAudioBuffer: AudioBuffer | null = null;
let outputStarted = false;
if (audioData) { if (audioData) {
updateProgress('processing', 'Detecting supported audio codec...', 8); updateProgress('processing', 'Detecting supported audio codec...', 8);
@ -308,7 +310,9 @@ export async function stitchVideosAsync(
} }
await output.start(); await output.start();
outputStarted = true;
try {
// Track the highest timestamp we've written to ensure monotonicity // Track the highest timestamp we've written to ensure monotonicity
// Start at -frameInterval so first frame can be at timestamp 0 // Start at -frameInterval so first frame can be at timestamp 0
const frameInterval = 1 / MAX_OUTPUT_FPS; const frameInterval = 1 / MAX_OUTPUT_FPS;
@ -370,14 +374,14 @@ export async function stitchVideosAsync(
// Skip duplicate frames that land on the same timestamp slot // Skip duplicate frames that land on the same timestamp slot
// This ensures strict 60fps without exceeding the target rate // This ensures strict 60fps without exceeding the target rate
if (snappedTimestamp <= highestWrittenTimestamp) { if (snappedTimestamp < highestWrittenTimestamp) {
sample.close(); sample.close();
continue; continue;
} }
sample.setTimestamp(snappedTimestamp); sample.setTimestamp(snappedTimestamp);
sample.setDuration(frameInterval); sample.setDuration(frameInterval);
await videoSource.add(sample); await videoSource!.add(sample);
// Update highest written timestamp // Update highest written timestamp
highestWrittenTimestamp = snappedTimestamp; highestWrittenTimestamp = snappedTimestamp;
@ -418,14 +422,18 @@ export async function stitchVideosAsync(
const trimmedBuffer = trimAudioBuffer(pendingAudioBuffer, videoDuration); const trimmedBuffer = trimAudioBuffer(pendingAudioBuffer, videoDuration);
await audioSource.add(trimmedBuffer); await audioSource.add(trimmedBuffer);
await audioSource.close(); await audioSource.close();
audioSource = null; // Already closed
} }
// Flush encoder before finalizing container // 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); updateProgress('processing', 'Finalizing stitched video...', 97);
// Finalize output // Finalize output
await output.finalize(); await output!.finalize();
outputStarted = false; // Successfully finalized
output = null;
const buffer = bufferTarget.buffer; const buffer = bufferTarget.buffer;
if (!buffer) { if (!buffer) {
@ -441,6 +449,31 @@ export async function stitchVideosAsync(
); );
return outputBlob; 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) { } catch (error) {
const normalizedError = const normalizedError =
error instanceof Error ? error : new Error(String(error)); error instanceof Error ? error : new Error(String(error));

10
src/store/workflowStore.ts

@ -2938,7 +2938,7 @@ export const useWorkflowStore = create<WorkflowStore>((set, get) => ({
error: "Browser does not support video encoding", error: "Browser does not support video encoding",
progress: 0, progress: 0,
}); });
set({ isRunning: false, currentNodeId: null }); set({ isRunning: false, currentNodeIds: [] });
await logger.endSession(); await logger.endSession();
return; return;
} }
@ -2954,7 +2954,7 @@ export const useWorkflowStore = create<WorkflowStore>((set, get) => ({
error: "Need at least 2 video clips to stitch", error: "Need at least 2 video clips to stitch",
progress: 0, progress: 0,
}); });
set({ isRunning: false, currentNodeId: null }); set({ isRunning: false, currentNodeIds: [] });
await logger.endSession(); await logger.endSession();
return; return;
} }
@ -3018,7 +3018,7 @@ export const useWorkflowStore = create<WorkflowStore>((set, get) => ({
error: "Browser does not support video encoding", error: "Browser does not support video encoding",
progress: 0, progress: 0,
}); });
set({ isRunning: false, currentNodeId: null }); set({ isRunning: false, currentNodeIds: [] });
await logger.endSession(); await logger.endSession();
return; return;
} }
@ -3051,7 +3051,7 @@ export const useWorkflowStore = create<WorkflowStore>((set, get) => ({
error: "Connect a video input to apply ease curve", error: "Connect a video input to apply ease curve",
progress: 0, progress: 0,
}); });
set({ isRunning: false, currentNodeId: null }); set({ isRunning: false, currentNodeIds: [] });
await logger.endSession(); await logger.endSession();
return; return;
} }
@ -3127,7 +3127,7 @@ export const useWorkflowStore = create<WorkflowStore>((set, get) => ({
}); });
} }
set({ isRunning: false, currentNodeId: null }); set({ isRunning: false, currentNodeIds: [] });
await logger.endSession(); await logger.endSession();
return; return;
} }

39
src/utils/deduplicatedFetch.ts

@ -13,8 +13,8 @@
// Map of URL -> in-flight promise // Map of URL -> in-flight promise
const inFlightRequests = new Map<string, Promise<Response>>(); const inFlightRequests = new Map<string, Promise<Response>>();
// Map of URL -> cloned response data (since Response body can only be read once) // Map of URL -> cloned response metadata (since Response body can only be read once)
const responseCache = new Map<string, { data: unknown; timestamp: number }>(); const responseCache = new Map<string, { status: number; headers: Record<string, string>; bodyText: string; timestamp: number }>();
// Cache TTL in milliseconds (5 seconds - short enough to get fresh data, long enough to dedupe) // Cache TTL in milliseconds (5 seconds - short enough to get fresh data, long enough to dedupe)
const CACHE_TTL = 5000; const CACHE_TTL = 5000;
@ -59,10 +59,10 @@ export async function deduplicatedFetch(
// Check if we have a recent cached response // Check if we have a recent cached response
const cached = responseCache.get(cacheKey); const cached = responseCache.get(cacheKey);
if (cached && Date.now() - cached.timestamp < CACHE_TTL) { if (cached && Date.now() - cached.timestamp < CACHE_TTL) {
// Return a synthetic response with the cached data // Return a synthetic response with the cached metadata
return new Response(JSON.stringify(cached.data), { return new Response(cached.bodyText, {
status: 200, status: cached.status,
headers: { "Content-Type": "application/json" }, headers: cached.headers,
}); });
} }
@ -70,13 +70,13 @@ export async function deduplicatedFetch(
const existingRequest = inFlightRequests.get(cacheKey); const existingRequest = inFlightRequests.get(cacheKey);
if (existingRequest) { if (existingRequest) {
// Wait for the existing request and clone it for this caller // 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 // The response body may have been consumed, so we rely on the cache
const cachedData = responseCache.get(cacheKey); const cachedData = responseCache.get(cacheKey);
if (cachedData) { if (cachedData) {
return new Response(JSON.stringify(cachedData.data), { return new Response(cachedData.bodyText, {
status: response.status, status: cachedData.status,
headers: { "Content-Type": "application/json" }, headers: cachedData.headers,
}); });
} }
// Fallback: this shouldn't happen, but return an error response // Fallback: this shouldn't happen, but return an error response
@ -88,14 +88,17 @@ export async function deduplicatedFetch(
// Create new request // Create new request
const requestPromise = fetch(url, options) const requestPromise = fetch(url, options)
.then(async (response) => { .then(async (response) => {
// Clone and cache the response data before returning // Clone and cache the full response metadata before returning
if (response.ok) { try {
try { const bodyText = await response.clone().text();
const data = await response.clone().json(); responseCache.set(cacheKey, {
responseCache.set(cacheKey, { data, timestamp: Date.now() }); status: response.status,
} catch { headers: Object.fromEntries(response.headers.entries()),
// Response wasn't JSON, that's fine bodyText,
} timestamp: Date.now(),
});
} catch {
// Failed to read response body for caching
} }
return response; return response;
}) })

56
src/utils/imageStorage.ts

@ -295,6 +295,9 @@ async function externalizeNodeImages(
} as WorkflowNode; } as WorkflowNode;
} }
// In-flight saves guard to prevent duplicate concurrent uploads of the same image
const inFlightSaves = new Map<string, Promise<string>>();
/** /**
* Save an image and return its ID (with deduplication) * Save an image and return its ID (with deduplication)
* @param folder - "inputs" for user-uploaded images, "generations" for AI-generated images * @param folder - "inputs" for user-uploaded images, "generations" for AI-generated images
@ -317,31 +320,50 @@ async function saveImageAndGetId(
return savedImageIds.get(hash)!; 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 // Use existing ID if provided (for consistency with imageHistory), otherwise generate new
const imageId = existingId || generateImageId(); const imageId = existingId || generateImageId();
const response = await fetchWithTimeout( const savePromise = (async () => {
"/api/workflow-images", const response = await fetchWithTimeout(
{ "/api/workflow-images",
method: "POST", {
headers: { "Content-Type": "application/json" }, method: "POST",
body: JSON.stringify({ headers: { "Content-Type": "application/json" },
workflowPath, body: JSON.stringify({
imageId, workflowPath,
imageData, imageId,
folder, 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) { if (!existingId) {
throw new Error(`Failed to save image: ${result.error}`); inFlightSaves.set(hash, savePromise);
} }
savedImageIds.set(hash, imageId); try {
return imageId; return await savePromise;
} catch (error) {
throw error;
} finally {
inFlightSaves.delete(hash);
}
} }
/** /**

Loading…
Cancel
Save