From 85441aa63780925e8feee3f8dd1bcf8b97011cce Mon Sep 17 00:00:00 2001 From: Vito Sansevero Date: Sat, 31 Oct 2026 23:13:59 -0700 Subject: [PATCH] fix: address review-bot findings on merged PRs (#593) Squashed from PR #595. Closes #593. --- __init__.py | 7 ++- tests/js/prompt-sender.test.js | 48 +++++++++++++++++- tests/test_maintenance_pair_metadata.py | 4 +- tests/test_native_memory.py | 8 +-- tests/test_saved_instructions.py | 12 +++-- tests/test_utils_parallel.py | 67 ++++++++++++++++++++++++- utils/image_monitor.py | 5 +- utils/native_memory.py | 10 ++-- utils/parallel.py | 10 ++-- web/js/workspace/prompt-sender.mjs | 34 ++++++------- 10 files changed, 165 insertions(+), 40 deletions(-) diff --git a/__init__.py b/__init__.py index b269c00f..d95dfff1 100644 --- a/__init__.py +++ b/__init__.py @@ -448,8 +448,11 @@ except Exception as e: __all__ = ["NODE_CLASS_MAPPINGS", "NODE_DISPLAY_NAME_MAPPINGS", "WEB_DIRECTORY"] -# One startup line. ASCII only: a Windows console using a legacy code page -# cannot print emoji, and a UnicodeEncodeError here would abort loading the node. +# One startup line, printed at import on purpose: it is the package's banner on +# stdout, as ComfyUI's other custom nodes print theirs. Not a log record: the +# console log handler hides INFO by default, so a logger would drop it. ASCII +# only: a Windows console using a legacy code page cannot print emoji, and a +# UnicodeEncodeError here would abort loading the node. print( f"\033[94m[ComfyUI-PromptManager] Version:\033[0m {get_version()}" f" - {len(NODE_CLASS_MAPPINGS)} tools loaded" diff --git a/tests/js/prompt-sender.test.js b/tests/js/prompt-sender.test.js index 3e74eb8c..e5b3835d 100644 --- a/tests/js/prompt-sender.test.js +++ b/tests/js/prompt-sender.test.js @@ -255,7 +255,8 @@ test('All is offered only with two or more eligible nodes and is never selected test('Apply to All previews every node first, then sends one after another in node id order',async()=>{ const {createSendSession,ALL_NODES}=await source;const s=await allSession(createSendSession); s.session.selectTarget(ALL_NODES);await s.session.apply(); - assert.deepEqual(order(s.calls),['preview 2','preview 9','preview 10','apply 2','preview 9','apply 9','preview 10','apply 10']); + // Each node takes a fresh ticket, the first too: the host keeps only its latest 100 (#593). + assert.deepEqual(order(s.calls),['preview 2','preview 9','preview 10','preview 2','apply 2','preview 9','apply 9','preview 10','apply 10']); assert.equal(s.session.state.phase,'success');assert.equal(s.session.state.message,'Sent to 3 of 3 nodes in Flow - ComfyUI.'); await s.session.apply();assert.equal(s.calls.filter(call=>call[0]==='apply').length,3);s.session.dispose(); }); @@ -363,3 +364,48 @@ test('a changed prepend or append after the review refuses that node',async()=>{ assert.deepEqual(s.calls.filter(call=>call[0]==='apply').map(call=>call[1]),['2','10']); assert.match(s.session.state.message,/^Sent to 2 of 3 nodes in Flow - ComfyUI\. #9 Node 9: Its prompt changed since the preview\./);s.session.dispose(); }); + +// #593: Send to All through the real Toolkit bridge and the real adapter, as the review bot reproduced it. +function realWindow(count) { + const listeners=new Set(); + const channelFactory=()=>{const own=new Set();return{ + addEventListener(type,listener){if(type==='message'){own.add(listener);listeners.add(listener);}},removeEventListener(type,listener){listeners.delete(listener);}, + postMessage(data){for(const listener of [...listeners])if(!own.has(listener))queueMicrotask(()=>listener({data:structuredClone(data),origin:'https://comfy.example'}));},close(){}, + };}; + // Like LiteGraph, a widget write moves the version of the graph it is in. + const graph=()=>{const made={_version:1,setDirtyCanvas(){made._version++;}};made._nodes=Array.from({length:count},(_,index)=>{ + const widgets=['text','prepend_text','append_text'].map(name=>({name,value:name==='text'?'old':'',inputEl:{value:name==='text'?'old':''}})); + return {id:index+1,type:'PromptManager',title:`Node ${index+1}`,widgets,widgets_values:widgets.map(widget=>widget.value),inputs:[],setDirtyCanvas(){}}; + });return made;}; + return {channelFactory,graph}; +} +async function realSend(count,{afterFirstApply}={}) { + const {createSendSession,ALL_NODES}=await source; + const {createBridgeHost,createBridgeClient}=await import('../../web/js/shared/toolkit/prompt-bridge.js'); + const {createPromptAdapter}=await import('../../web/js/prompt-bridge-adapter.js'); + const win=realWindow(count);const app={graph:win.graph(),canvas:{}}; + const adapter=createPromptAdapter(app,{runtime:{}});let applied=0; + const watched={...adapter,apply(snapshot){const result=adapter.apply(snapshot);if(++applied===1)afterFirstApply?.(app,win);return result;}}; + const options={namespace:'/prompt_manager/',origin:'https://comfy.example',channelFactory:win.channelFactory}; + const host=createBridgeHost({...options,label:'Flow - ComfyUI',adapter:watched}); + const client=createBridgeClient(options); + const session=createSendSession({client,roles:{positive:'new'},onChange:()=>{}}); + try { + await session.discover();session.selectTarget(ALL_NODES);await session.apply(); + return {state:session.state,app}; + } finally { session.dispose();host.dispose(); } +} +test('All refuses the later nodes when another workflow is loaded after the first send (#593)',async()=>{ + let first; + const {state,app}=await realSend(3,{afterFirstApply:(target,win)=>{first=target.graph;target.graph=win.graph();target.graph._version=first._version;}}); + // The new workflow has the same ids, text and composition, and even the same graph version: only node identity differs. + assert.equal(first._nodes[0].widgets[0].value,'new'); + assert.deepEqual(app.graph._nodes.map(node=>node.widgets[0].value),['old','old','old']); + assert.equal(state.phase,'error'); + assert.match(state.message,/^Sent to 1 of 3 nodes in Flow - ComfyUI\. #2 Node 2: The workflow changed since this window was found\./); +}); +test('All sends to every node of a window with more than 100 of them (#593)',async()=>{ + const {state,app}=await realSend(101); + assert.equal(state.message,'Sent to 101 of 101 nodes in Flow - ComfyUI.'); + assert.ok(app.graph._nodes.every(node=>node.widgets[0].value==='new')); +}); diff --git a/tests/test_maintenance_pair_metadata.py b/tests/test_maintenance_pair_metadata.py index df401098..ed76f070 100644 --- a/tests/test_maintenance_pair_metadata.py +++ b/tests/test_maintenance_pair_metadata.py @@ -171,7 +171,7 @@ class TestClassification(PairMetadataCase): self.assertEqual(self.preview().targets, [[prompt, True, True]]) def test_category_is_case_sensitive(self): - _a, prompt = self.pair("case", "n1", category="Portrait") + self.pair("case", "n1", category="Portrait") self.pair("case", "n2", category="portrait") draft = self.preview() self.assertEqual(draft.targets, []) @@ -292,7 +292,7 @@ class TestClassification(PairMetadataCase): ) def test_protected_rows_come_first_within_fifty(self): - _c, conflict = self.pair("zz conflict", "n1", category="x") + self.pair("zz conflict", "n1", category="x") self.pair("zz conflict", "n2", category="y") for n in range(60): self.pair(f"fill {n:02d}", category="portrait") diff --git a/tests/test_native_memory.py b/tests/test_native_memory.py index c1892a33..06a04d0d 100644 --- a/tests/test_native_memory.py +++ b/tests/test_native_memory.py @@ -2,7 +2,7 @@ import sys import unittest -from unittest import mock +import unittest.mock as mock from utils import native_memory @@ -10,7 +10,9 @@ from utils import native_memory class ReleaseFreeMemoryTests(unittest.TestCase): def setUp(self): # Each test looks the symbol up again. - patcher = mock.patch.multiple(native_memory, _trim=None, _looked_up=False) + patcher = mock.patch.object( + native_memory, "_trim", native_memory._NOT_LOOKED_UP + ) patcher.start() self.addCleanup(patcher.stop) @@ -22,7 +24,7 @@ class ReleaseFreeMemoryTests(unittest.TestCase): def test_other_platforms_do_nothing(self): for platform in ("win32", "darwin", "cygwin"): with self.subTest(platform), mock.patch.object(sys, "platform", platform): - native_memory._looked_up = False + native_memory._trim = native_memory._NOT_LOOKED_UP with mock.patch.object(native_memory.ctypes, "CDLL") as cdll: self.assertFalse(native_memory.release_free_memory()) cdll.assert_not_called() diff --git a/tests/test_saved_instructions.py b/tests/test_saved_instructions.py index f7f4b09f..e32011d7 100644 --- a/tests/test_saved_instructions.py +++ b/tests/test_saved_instructions.py @@ -153,7 +153,7 @@ class ConfigTestCase(unittest.TestCase): path = os.path.join(tmpdir, "config.json") with open(path, "w", encoding="utf-8") as handle: json.dump({"autotag": {"defaults": defaults}}, handle) - before = os.stat(path).st_mtime_ns, open(path, "rb").read() + before = os.stat(path).st_mtime_ns, self.read_bytes(path) if warnings: with self.assertLogs("prompt_manager.config", level="WARNING") as logs: PromptManagerConfig.load_from_file(path) @@ -162,9 +162,13 @@ class ConfigTestCase(unittest.TestCase): else: PromptManagerConfig.load_from_file(path) # Nothing is written at load. - self.assertEqual( - (os.stat(path).st_mtime_ns, open(path, "rb").read()), before - ) + self.assertEqual((os.stat(path).st_mtime_ns, self.read_bytes(path)), before) + + @staticmethod + def read_bytes(path): + # Closed at once: Windows cannot delete the temp dir with it open. + with open(path, "rb") as handle: + return handle.read() def state(self): return [dict(e) for e in AutotagConfig.INSTRUCTIONS], dict( diff --git a/tests/test_utils_parallel.py b/tests/test_utils_parallel.py index 21fc717c..4d058688 100644 --- a/tests/test_utils_parallel.py +++ b/tests/test_utils_parallel.py @@ -12,10 +12,12 @@ import unittest sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) -from utils import parallel # noqa: E402 +from utils import exit_signal, parallel # noqa: E402 from utils.parallel import JOB_THREAD_PREFIX, map_parallel # noqa: E402 WAIT = 30 +# A map that has not returned by then is stuck (#593). +HANG_LIMIT = 5 def _job_threads(): @@ -163,5 +165,68 @@ class TestMapParallel(unittest.TestCase): map_parallel(boom, [1, 2, 3], workers=3) +class TestErrorAfterShutdownCancel(unittest.TestCase): + """#593: ``shutdown_job_pool`` cancels the call's queued items without + notifying them, so the error path must not wait on them (exit hung).""" + + def setUp(self): + exit_signal.reset() + self.addCleanup(exit_signal.reset) + self.assertTrue(parallel.shutdown_job_pool(WAIT)) + self.addCleanup(parallel.shutdown_job_pool, WAIT) + + def _run_in_thread(self, func, items): + """``map_parallel`` on a daemon thread: (thread, outcome list), so a + hang fails the test instead of hanging the suite.""" + outcome = [] + + def call(): + try: + outcome.append(("result", map_parallel(func, items, 1))) + except Exception as error: # the outcome the test checks + outcome.append(("error", error)) + + thread = threading.Thread(target=call, daemon=True) + thread.start() + return thread, outcome + + def _hold_first_then_cancel_rest(self, first): + """One worker runs item 0 (``first``) while items 1 and 2 queue; + shutdown cancels the queued two, then item 0 is released.""" + entered, release = threading.Event(), threading.Event() + + def item(n): + if n == 0: + entered.set() + self.assertTrue(release.wait(WAIT)) + return first() + return n + + thread, outcome = self._run_in_thread(item, [0, 1, 2]) + self.assertTrue(entered.wait(WAIT)) + self.assertFalse(parallel.shutdown_job_pool(0)) # item 0 still runs + release.set() + thread.join(HANG_LIMIT) + self.assertFalse(thread.is_alive(), "map_parallel hung on cancelled items") + return outcome + + def test_a_running_item_raising_after_shutdown_cancel_returns(self): + def fail(): + raise ValueError("zero") + + outcome = self._hold_first_then_cancel_rest(fail) + self.assertEqual(outcome[0][0], "error") + self.assertIsInstance(outcome[0][1], ValueError) + + def test_exit_requested_inline_after_shutdown_cancel_returns(self): + def signal_exit(): + exit_signal.request() # the cancelled items then run inline + return 0 + + outcome = self._hold_first_then_cancel_rest(signal_exit) + self.assertEqual(outcome[0][0], "error") + self.assertIsInstance(outcome[0][1], exit_signal.ExitRequested) + + if __name__ == "__main__": unittest.main() diff --git a/utils/image_monitor.py b/utils/image_monitor.py index eb4c030a..51a8581e 100644 --- a/utils/image_monitor.py +++ b/utils/image_monitor.py @@ -113,7 +113,10 @@ def _start_daemon_observer(observer) -> None: def start() -> None: try: observer.start() - except BaseException as e: # re-raised on the caller's thread + # BaseException on purpose: a KeyboardInterrupt or SystemExit raised + # by start() is not lost on this thread; _start_daemon_observer + # re-raises it on the caller's thread below. + except BaseException as e: raised.append(e) starter = threading.Thread( diff --git a/utils/native_memory.py b/utils/native_memory.py index c7f7c3b0..24aba8b1 100644 --- a/utils/native_memory.py +++ b/utils/native_memory.py @@ -16,8 +16,9 @@ import sys import threading _lock = threading.Lock() -_trim = None -_looked_up = False +# Not looked up yet; after the lookup, the function or None. +_NOT_LOOKED_UP = object() +_trim = _NOT_LOOKED_UP def _glibc() -> bool: @@ -33,10 +34,9 @@ def _glibc() -> bool: def _malloc_trim(): """glibc's ``malloc_trim``, looked up once; None where there is none.""" - global _trim, _looked_up + global _trim with _lock: - if not _looked_up: - _looked_up = True + if _trim is _NOT_LOOKED_UP: _trim = _lookup() return _trim diff --git a/utils/parallel.py b/utils/parallel.py index 99094224..2798839b 100644 --- a/utils/parallel.py +++ b/utils/parallel.py @@ -165,10 +165,12 @@ def _collect(func, items, futures): and wait for its running ones, as the old per-call pool's exit did.""" try: return _run_rest(func, items, futures) - except BaseException: - for future in futures: - future.cancel() - wait(futures) + except BaseException: # re-raised below, after the running items stop + # Wait only for what could not be cancelled (running or done). A + # future ``shutdown(cancel_futures=True)`` already cancelled stays + # CANCELLED, never notified, and ``wait`` would block on it forever + # (#593); ``cancel()`` is True for those as well. + wait([future for future in futures if not future.cancel()]) raise diff --git a/web/js/workspace/prompt-sender.mjs b/web/js/workspace/prompt-sender.mjs index 196ed480..5572c93f 100644 --- a/web/js/workspace/prompt-sender.mjs +++ b/web/js/workspace/prompt-sender.mjs @@ -63,15 +63,19 @@ export function createSendSession({client, roles:rawRoles, onChange}) { const target=host.targets.find(item=>item.id===state.targetId); return target ? [target] : []; } - // checkRevision is off only for a later node of "All", whose revision our own earlier write may have moved. - async function fetchPreview(request,hostId,target,checkRevision=true) { + // The adapter's revision is ":". Loading a workflow makes new node objects, so a new + // identity; our own write moves only the graph version. A string without the separator is all identity. + const nodeIdentity = revision => typeof revision==='string' && revision.includes(':') ? revision.slice(0,revision.indexOf(':')) : revision; + // graphMayMove is on only for a later node of "All", whose graph version our own earlier write may have moved; the node + // identity is still checked, so a write never lands in another workflow loaded in between (#593). + async function fetchPreview(request,hostId,target,graphMayMove=false) { const preview=await client.preview(hostId,target.id,roles,{signal:request.signal}); if (!current(request)) return null; const fields=preview?.fields; if(typeof preview?.ticket!=='string' || !preview.ticket || preview.target?.id!==target.id || !Array.isArray(fields) || fields.length!==Object.keys(roles).length || new Set(fields.map(field=>field?.role)).size!==fields.length || Array.from(fields).some(field=>!field || !Object.hasOwn(roles,field.role) || typeof field.before!=='string' || field.after!==roles[field.role])) throw Error('ComfyUI returned an unexpected preview. Find the destination and preview again.'); // Undefined on a tab still running older code, which has nothing to compare. const found=target.revision; - if(checkRevision && found!==undefined && preview.revision!==found) throw Error('The workflow changed since this window was found. Nothing was sent. Find windows again.'); + if(found!==undefined && (graphMayMove ? nodeIdentity(preview.revision)!==nodeIdentity(found) : preview.revision!==found)) throw Error('The workflow changed since this window was found. Nothing was sent. Find windows again.'); return preview; } function checkAcknowledgement(acknowledgement,targetId) { @@ -95,25 +99,21 @@ export function createSendSession({client, roles:rawRoles, onChange}) { return reviewed; } /** - * Sends the reviewed nodes one after another, best effort past refusals. Only the first node of a pass taken in this - * click keeps its ticket: a held review may have expired at the host (30 s), and our own writes may make later - * tickets stale, so every other node takes a fresh ticket that must show exactly the reviewed text. + * Sends the reviewed nodes one after another, best effort past refusals. Every node takes a fresh ticket that must + * show exactly the reviewed text: a held review may have expired at the host (30 s), our own writes make later + * tickets stale, and the host keeps only its latest 100 tickets, so a pass over more nodes has evicted the first (#593). */ - async function applyEach(request,hostId,reviewed,where,held) { + async function applyEach(request,hostId,reviewed,where) { const sent=[],refused=[];let written=false; for (const [index,{target,preview}] of reviewed.entries()) { let posted=false; try { - let ticket=preview.ticket; - if (index>0 || held) { - // Until something was posted the revision is still the discovered one; after that our own write may move it. - const fresh=await fetchPreview(request,hostId,target,!written); - if (!fresh) return; - if (!sameChanges(preview,fresh)) throw Error('Its prompt changed since the preview. Preview again.'); - ticket=fresh.ticket; - } + // Until something was posted the revision is still the discovered one; after that our own write may move it. + const fresh=await fetchPreview(request,hostId,target,written); + if (!fresh) return; + if (!sameChanges(preview,fresh)) throw Error('Its prompt changed since the preview. Preview again.'); posted=written=true; - const acknowledgement=await client.apply(hostId,ticket,{signal:request.signal}); + const acknowledgement=await client.apply(hostId,fresh.ticket,{signal:request.signal}); if (!current(request)) return; checkAcknowledgement(acknowledgement,target.id); sent.push(target); @@ -191,7 +191,7 @@ export function createSendSession({client, roles:rawRoles, onChange}) { let reviewed; try { reviewed=held?.length ? held : await previewEach(request,hostId,targets); } catch(error) { if(current(request)) emit({phase:'error',message:error.message}); return; } - if (reviewed) await applyEach(request,hostId,reviewed,windowLabel,Boolean(held?.length)); + if (reviewed) await applyEach(request,hostId,reviewed,windowLabel); return; } const [target]=targets;