@@ -2164,6 +2164,72 @@ export class AutomationEngine implements IAutomationService {
21642164 await this . releaseSuspension ( run , reason ) ;
21652165 }
21662166
2167+ /**
2168+ * [#15832] Drop a map entry for a run this process has a store-authoritative
2169+ * per-id "no row" answer for — a run parked HERE and consumed by another
2170+ * replica.
2171+ *
2172+ * ## Why this exists at all, and why not on the `'lost'` branch
2173+ *
2174+ * {@link forgetSuspendedRun} is the one eviction site, and it runs in
2175+ * whichever process CONSUMES the suspension. Put two replicas over one
2176+ * store and the parking process is routinely not that one: A parks a run
2177+ * and B resumes it, so A's entry is never removed by anything. The card
2178+ * that found this located the leak on `resumeInternal`'s `'lost'` branch,
2179+ * which returns before that choke point — but the NO-RACE shape leaks
2180+ * identically (A parks, only B ever resumes, A never attempts a claim and
2181+ * there is no `'lost'` at all), so an eviction hung on `'lost'` alone would
2182+ * leave the ordinary multi-replica deployment untouched.
2183+ *
2184+ * The retained snapshot is not only memory. Two readers hand it back:
2185+ * {@link listSuspendedRuns} — synchronous, cache-only, and the one listing
2186+ * on the `AutomationService` spec contract — and
2187+ * {@link listSuspendedRunsDurable}, which deliberately appends map entries
2188+ * the durable list lacks. After the other replica COMPLETES the run both
2189+ * report a phantom: a finished run listed as suspended, whose
2190+ * {@link getSuspendedScreen} answers `null`, so a consumer that lists and
2191+ * then opens gets an entry it cannot act on.
2192+ *
2193+ * ## What this does NOT do
2194+ *
2195+ * ⛔ It does not touch the cache-only listing's contract. The spec says
2196+ * `listSuspendedRuns()` lists "the currently suspended (paused) runs
2197+ * awaiting a resume"; this engine's own docblock adds only that it may
2198+ * OMIT runs (those parked in a previous process lifetime) because it reads
2199+ * the cache alone. Removing an entry therefore moves nothing: under-
2200+ * reporting is already inside that declared latitude, and over-reporting
2201+ * was never inside the promise. Nothing here makes either listing
2202+ * store-backed.
2203+ *
2204+ * ⛔ It does not notify the paused node's executor
2205+ * ({@link NodeExecutor.onSuspensionReleased}). That notification belongs to
2206+ * {@link forgetSuspendedRun} because it is the choke point every
2207+ * CONSUMPTION passes through, and an eviction is not a consumption — this
2208+ * process consumed nothing, the replica that did fired its own. Firing one
2209+ * here would tear down a pause twice, once per replica.
2210+ *
2211+ * ## The two guards, and why each is load-bearing
2212+ *
2213+ * - **no store** — the map IS the authority (`loadSuspendedRunStrict`
2214+ * returns from it directly), so there is no second reader to be wrong
2215+ * about and nothing may be dropped.
2216+ * - **{@link cacheOnlySuspensions}** — a run whose durable save failed was
2217+ * never handed to the store, so the store's "no row" is SILENCE about it
2218+ * rather than an answer (#13617). Evicting on that would convert
2219+ * {@link persistSuspendedRun}'s documented degradation — a failed save
2220+ * costs cross-restart durability, not in-process resumability — into a
2221+ * run that vanishes from its own process.
2222+ *
2223+ * A store read that THROWS must never reach here: an outage means the
2224+ * run's existence is UNKNOWN, not "gone". Every caller below is on a path
2225+ * where the store answered.
2226+ */
2227+ private evictConsumedSuspension ( runId : string ) : void {
2228+ if ( ! this . store ) return ;
2229+ if ( this . cacheOnlySuspensions . has ( runId ) ) return ;
2230+ this . suspendedRuns . delete ( runId ) ;
2231+ }
2232+
21672233 /**
21682234 * [#14333] Claim the right to advance this run past the node it is parked
21692235 * at — the CROSS-REPLICA half of the resume idempotency guard.
@@ -4996,6 +5062,13 @@ export class AutomationEngine implements IAutomationService {
49965062 // deliberately keeps such a run resumable in-process (it reports the
49975063 // lost durability at `error`).
49985064 if ( this . cacheOnlySuspensions . has ( runId ) ) return this . suspendedRuns . get ( runId ) ?? null ;
5065+ // [#15832] The store ANSWERED, and the answer is "no row". Any entry
5066+ // this process still holds for that run is a run it parked and another
5067+ // replica consumed — the phantom the two listings hand back. This is
5068+ // the definitive per-id evidence the paragraph above already rests on,
5069+ // so the same reading that refuses to serve it here stops publishing it
5070+ // there. A store read that threw never reaches this line.
5071+ this . evictConsumedSuspension ( runId ) ;
49995072 return null ;
50005073 }
50015074
@@ -5374,6 +5447,14 @@ export class AutomationEngine implements IAutomationService {
53745447 // already maps it to 409. A distinct code would be vocabulary
53755448 // nothing reads — add one the day a caller needs the
53765449 // difference.
5450+ // [#15832] The store's compare-and-set ANSWERED: no row is parked
5451+ // where this replica read it. Whatever snapshot this process
5452+ // still holds for the run is stale by construction — it names a
5453+ // node the run has left — so it stops being published by the two
5454+ // listings. ⛔ NOT `forgetSuspendedRun`: nothing was consumed
5455+ // here, and firing that choke point would tear the pause down a
5456+ // second time in this process on top of the winner's own.
5457+ this . evictConsumedSuspension ( runId ) ;
53775458 return {
53785459 success : false ,
53795460 code : 'RESUME_IN_PROGRESS' ,
@@ -6536,11 +6617,16 @@ export class AutomationEngine implements IAutomationService {
65366617 */
65376618 async listSuspendedRunsDurable ( ) : Promise < Array < { runId : string ; flowName : string ; nodeId : string ; correlation ?: string } > > {
65386619 const byId = new Map < string , { runId : string ; flowName : string ; nodeId : string ; correlation ?: string } > ( ) ;
6620+ // [#15832] Did the ENUMERATION answer? The reconcile below is allowed
6621+ // only when it did — see the merge comment for why a failed listing is
6622+ // silence rather than evidence.
6623+ let enumerated = false ;
65396624 if ( this . store ) {
65406625 try {
65416626 for ( const r of await this . store . list ( ) ) {
65426627 byId . set ( r . runId , { runId : r . runId , flowName : r . flowName , nodeId : r . nodeId , correlation : r . correlation } ) ;
65436628 }
6629+ enumerated = true ;
65446630 } catch ( err ) {
65456631 // #6299 — driver text to the structured slot, message one line,
65466632 // same as the two seams above. The SLOT differs: the `Logger`
@@ -6605,8 +6691,35 @@ export class AutomationEngine implements IAutomationService {
66056691 // only {@link cacheOnlySuspensions } answer out of the map. Applying that
66066692 // qualifier here would let a truncated or failed enumeration silently
66076693 // drop live runs from an operability listing.
6608- for ( const r of this . suspendedRuns . values ( ) ) {
6694+ //
6695+ // [#15832] What that reasoning leaves open is a run this process parked
6696+ // and ANOTHER replica has since consumed: absent from the durable list
6697+ // because it is finished, appended here, and published as suspended by
6698+ // a listing that also backs the cache-only one. The paragraph above is
6699+ // right that list-absence is not evidence — so this asks for the
6700+ // evidence instead. `store.load` is the same definitive per-id read
6701+ // {@link loadSuspendedRunStrict } rests on, and it is bought only for the
6702+ // entries that look suspicious: a healthy process, whose map entries all
6703+ // appear in the durable list, buys none. A read that THROWS leaves the
6704+ // entry standing (unknown is not gone), and a store that could not be
6705+ // enumerated at all is not probed row by row — an outage would answer
6706+ // for every live run in the process.
6707+ for ( const r of [ ...this . suspendedRuns . values ( ) ] ) {
66096708 if ( byId . has ( r . runId ) ) continue ;
6709+ if ( enumerated && ! this . cacheOnlySuspensions . has ( r . runId ) ) {
6710+ let stored : SuspendedRun | null ;
6711+ try {
6712+ stored = await this . store ! . load ( r . runId ) ;
6713+ } catch {
6714+ // Unknown, not gone — keep the entry and publish it, exactly
6715+ // as this method did before the reconcile existed.
6716+ stored = r ;
6717+ }
6718+ if ( stored === null ) {
6719+ this . evictConsumedSuspension ( r . runId ) ;
6720+ continue ;
6721+ }
6722+ }
66106723 byId . set ( r . runId , { runId : r . runId , flowName : r . flowName , nodeId : r . nodeId , correlation : r . correlation } ) ;
66116724 }
66126725 return [ ...byId . values ( ) ] ;
0 commit comments